From 8dc4d516d7731248cea3cebf89fecf14930890f4 Mon Sep 17 00:00:00 2001 From: "kaiyi.lk" Date: Thu, 28 Apr 2022 16:09:36 +0800 Subject: [PATCH] [ISSUE #3949] v2 support --- .../v2/service/cluster/ConsumerService.java | 3 +- .../v2/service/cluster/RouteServiceTest.java | 19 +- .../test/grpc/v1/ClusterGrpcTest.java | 123 ----------- .../rocketmq/test/grpc/v1/GrpcBaseTest.java | 207 ------------------ .../rocketmq/test/grpc/v2/GrpcBaseTest.java | 20 +- 5 files changed, 27 insertions(+), 345 deletions(-) delete mode 100644 test/src/test/java/org/apache/rocketmq/test/grpc/v1/ClusterGrpcTest.java delete mode 100644 test/src/test/java/org/apache/rocketmq/test/grpc/v1/GrpcBaseTest.java diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ConsumerService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ConsumerService.java index c5519dccee..8df9fd14f7 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ConsumerService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ConsumerService.java @@ -278,7 +278,8 @@ public class ConsumerService extends BaseService { String brokerAddr = this.getBrokerAddr(ctx, receiptHandle.getBrokerName()); ChangeInvisibleTimeRequestHeader requestHeader = convertToChangeInvisibleTimeRequestHeader(ctx, request); - future = this.writeConsumer.changeInvisibleTimeAsync(ctx, brokerAddr, receiptHandle.getBrokerName(), "", requestHeader) + future = this.writeConsumer.changeInvisibleTimeAsync(ctx, brokerAddr, receiptHandle.getBrokerName(), + request.getMessageId(), requestHeader) .thenApply(result -> convertToChangeInvisibleDurationResponse(ctx, request, result)); } catch (Throwable t) { future.completeExceptionally(t); diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/RouteServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/RouteServiceTest.java index b013eb5965..65c0889a67 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/RouteServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/RouteServiceTest.java @@ -61,16 +61,19 @@ public class RouteServiceTest extends BaseServiceTest { .setResourceNamespace(NAMESPACE) .build(); - private static final Settings WITH_HOST_SETTINGS = Settings.newBuilder() - .setAccessPoint(Endpoints.newBuilder() - .addAddresses(Address.newBuilder() - .setPort(80) - .setHost("host") - .build()) - .setScheme(AddressScheme.DOMAIN_NAME) + private static final Endpoints WITH_HOST_ENDPOINT = Endpoints.newBuilder() + .addAddresses(Address.newBuilder() + .setPort(80) + .setHost("host") .build()) + .setScheme(AddressScheme.DOMAIN_NAME) .build(); + private static final Settings WITH_HOST_SETTINGS = Settings.newBuilder() + .setAccessPoint(WITH_HOST_ENDPOINT) + .build(); + + private static final Settings INVALID_HOST_SETTINGS = Settings.newBuilder() .setAccessPoint(Endpoints.getDefaultInstance()) .build(); @@ -185,6 +188,7 @@ public class RouteServiceTest extends BaseServiceTest { .setTopic(Resource.newBuilder() .setName("topic") .build()) + .setEndpoints(WITH_HOST_ENDPOINT) .build()); QueryRouteResponse response = future.get(); @@ -228,6 +232,7 @@ public class RouteServiceTest extends BaseServiceTest { when(grpcClientManager.getClientSettings(any(Context.class))).thenReturn(WITH_HOST_SETTINGS); CompletableFuture future = routeService.queryAssignment(Context.current(), QueryAssignmentRequest.newBuilder() + .setEndpoints(WITH_HOST_ENDPOINT) .setTopic(Resource.newBuilder() .setName("topic") .build()) diff --git a/test/src/test/java/org/apache/rocketmq/test/grpc/v1/ClusterGrpcTest.java b/test/src/test/java/org/apache/rocketmq/test/grpc/v1/ClusterGrpcTest.java deleted file mode 100644 index 5d40e39e29..0000000000 --- a/test/src/test/java/org/apache/rocketmq/test/grpc/v1/ClusterGrpcTest.java +++ /dev/null @@ -1,123 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.rocketmq.test.grpc.v1; - -import apache.rocketmq.v1.AckMessageResponse; -import apache.rocketmq.v1.Address; -import apache.rocketmq.v1.AddressScheme; -import apache.rocketmq.v1.Endpoints; -import apache.rocketmq.v1.MessagingServiceGrpc; -import apache.rocketmq.v1.QueryRouteResponse; -import apache.rocketmq.v1.ReceiveMessageResponse; -import apache.rocketmq.v1.SendMessageResponse; -import com.google.common.base.Stopwatch; -import io.grpc.Channel; -import java.net.URL; -import java.util.concurrent.TimeUnit; -import org.apache.rocketmq.proxy.config.ConfigurationManager; -import org.apache.rocketmq.proxy.grpc.v1.GrpcMessagingProcessor; -import org.apache.rocketmq.proxy.grpc.v1.service.ClusterGrpcService; -import org.apache.rocketmq.proxy.grpc.v1.service.GrpcForwardService; -import org.junit.After; -import org.junit.Before; -import org.junit.Test; - -import static org.apache.rocketmq.common.message.MessageClientIDSetter.createUniqID; -import static org.apache.rocketmq.proxy.config.ConfigurationManager.RMQ_PROXY_HOME; -import static org.junit.Assert.assertTrue; - -public class ClusterGrpcTest extends GrpcBaseTest { - - private final int PORT = 8083; - private GrpcForwardService grpcForwardService; - private MessagingServiceGrpc.MessagingServiceBlockingStub blockingStub; - - @Before - public void setUp() throws Exception { - String mockProxyHome = "/mock/rmq/proxy/home"; - URL mockProxyHomeURL = getClass().getClassLoader().getResource("rmq-proxy-home"); - if (mockProxyHomeURL != null) { - mockProxyHome = mockProxyHomeURL.toURI().getPath(); - } - System.setProperty(RMQ_PROXY_HOME, mockProxyHome); - ConfigurationManager.initEnv(); - ConfigurationManager.intConfig(); - ConfigurationManager.getProxyConfig().setGrpcServerPort(PORT); - ConfigurationManager.getProxyConfig().setNameSrvAddr(nsAddr); - grpcForwardService = new ClusterGrpcService(); - grpcForwardService.start(); - GrpcMessagingProcessor processor = new GrpcMessagingProcessor(grpcForwardService); - Channel channel = setUpServer(processor, ConfigurationManager.getProxyConfig().getGrpcServerPort(), true); - blockingStub = MessagingServiceGrpc.newBlockingStub(channel); - } - - @After - public void tearDown() throws Exception { - grpcForwardService.shutdown(); - shutdown(); - } - - @Test - public void testQueryRoute() { - String topic = initTopic(); - QueryRouteResponse response = blockingStub.queryRoute(buildQueryRouteRequest(topic, Endpoints.newBuilder() - .setScheme(AddressScheme.IPv4) - .addAddresses(Address.newBuilder() - .setHost("127.0.0.1") - .setPort(PORT) - .build()) - .build())); - assertQueryRoute(response, brokerControllerList.size()); - } - - @Test - public void testSendReceiveMessage() { - String group = "group"; - String messageId = createUniqID(); - SendMessageResponse sendResponse = blockingStub.sendMessage(buildSendMessageRequest(broker1Name, messageId)); - assertSendMessage(sendResponse, messageId); - - ReceiveMessageResponse receiveResponse = blockingStub.withDeadlineAfter(3, TimeUnit.SECONDS) - .receiveMessage(buildReceiveMessageRequest(group, broker1Name)); - assertReceiveMessage(receiveResponse, messageId); - String receiptHandle = receiveResponse.getMessages(0).getSystemAttribute().getReceiptHandle(); - AckMessageResponse ackMessageResponse = blockingStub.ackMessage(buildAckMessageRequest(group, broker1Name, receiptHandle)); - assertAck(ackMessageResponse); - } - - @Test - public void testSendReceiveDelayMessage() { - String group = "group"; - String messageId = createUniqID(); - SendMessageResponse sendResponse = blockingStub.sendMessage(buildSendDelayMessageRequest(broker1Name, messageId, 2)); - assertSendMessage(sendResponse, messageId); - - Stopwatch stopwatch = Stopwatch.createStarted(); - ReceiveMessageResponse receiveResponse = blockingStub.withDeadlineAfter(10, TimeUnit.SECONDS) - .receiveMessage(buildReceiveMessageRequest(group, broker1Name)); - long rcvTime = stopwatch.elapsed(TimeUnit.SECONDS); - assertTrue(Math.abs(rcvTime - 5) < 2); - - assertReceiveMessage(receiveResponse, messageId); - String receiptHandle = receiveResponse.getMessages(0).getSystemAttribute().getReceiptHandle(); - AckMessageResponse ackMessageResponse = blockingStub.ackMessage(buildAckMessageRequest(group, broker1Name, receiptHandle)); - assertAck(ackMessageResponse); - } - - -} diff --git a/test/src/test/java/org/apache/rocketmq/test/grpc/v1/GrpcBaseTest.java b/test/src/test/java/org/apache/rocketmq/test/grpc/v1/GrpcBaseTest.java deleted file mode 100644 index 5494f5405d..0000000000 --- a/test/src/test/java/org/apache/rocketmq/test/grpc/v1/GrpcBaseTest.java +++ /dev/null @@ -1,207 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.rocketmq.test.grpc.v1; - -import apache.rocketmq.v1.AckMessageRequest; -import apache.rocketmq.v1.AckMessageResponse; -import apache.rocketmq.v1.Endpoints; -import apache.rocketmq.v1.Message; -import apache.rocketmq.v1.Partition; -import apache.rocketmq.v1.QueryRouteRequest; -import apache.rocketmq.v1.QueryRouteResponse; -import apache.rocketmq.v1.ReceiveMessageRequest; -import apache.rocketmq.v1.ReceiveMessageResponse; -import apache.rocketmq.v1.Resource; -import apache.rocketmq.v1.SendMessageRequest; -import apache.rocketmq.v1.SendMessageResponse; -import apache.rocketmq.v1.SystemAttribute; -import com.google.protobuf.ByteString; -import com.google.protobuf.Duration; -import com.google.protobuf.Timestamp; -import com.google.rpc.Code; -import io.grpc.Channel; -import io.grpc.ServerInterceptors; -import io.grpc.ServerServiceDefinition; -import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder; -import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder; -import io.grpc.netty.shaded.io.netty.handler.ssl.ApplicationProtocolConfig; -import io.grpc.netty.shaded.io.netty.handler.ssl.SslContextBuilder; -import io.grpc.netty.shaded.io.netty.handler.ssl.SslProvider; -import io.grpc.testing.GrpcCleanupRule; -import io.netty.handler.ssl.ApplicationProtocolNames; -import io.netty.handler.ssl.util.InsecureTrustManagerFactory; -import io.netty.handler.ssl.util.SelfSignedCertificate; -import java.io.IOException; -import java.security.cert.CertificateException; -import java.util.concurrent.TimeUnit; -import org.apache.rocketmq.proxy.config.ConfigurationManager; -import org.apache.rocketmq.proxy.grpc.interceptor.ContextInterceptor; -import org.apache.rocketmq.proxy.grpc.interceptor.HeaderInterceptor; -import org.apache.rocketmq.test.base.BaseConf; -import org.junit.Rule; - -import static org.assertj.core.api.Assertions.assertThat; - -public class GrpcBaseTest extends BaseConf { - /** - * This rule manages automatic graceful shutdown for the registered servers and channels at the end of test. - */ - @Rule - public final GrpcCleanupRule grpcCleanup = new GrpcCleanupRule(); - - private static final int defaultQueueNums = 8; - - protected Channel setUpServer(apache.rocketmq.v1.MessagingServiceGrpc.MessagingServiceImplBase serverImpl, - int port, boolean enableInterceptor) throws IOException, CertificateException { - SelfSignedCertificate selfSignedCertificate = new SelfSignedCertificate(); - ServerServiceDefinition serviceDefinition = ServerInterceptors.intercept(serverImpl); - if (enableInterceptor) { - serviceDefinition = ServerInterceptors.intercept(serverImpl, new ContextInterceptor(), new HeaderInterceptor()); - } - // Create a server, add service, start, and register for automatic graceful shutdown. - grpcCleanup.register(NettyServerBuilder.forPort(port) - .directExecutor() - .addService(serviceDefinition) - .useTransportSecurity(selfSignedCertificate.certificate(), selfSignedCertificate.privateKey()) - .build() - .start()); - // Create a client channel and register for automatic graceful shutdown. - return grpcCleanup.register(NettyChannelBuilder.forAddress("127.0.0.1", port) - .directExecutor() - .sslContext(SslContextBuilder - .forClient() - .sslProvider(SslProvider.OPENSSL) - .trustManager(InsecureTrustManagerFactory.INSTANCE) - .applicationProtocolConfig(new ApplicationProtocolConfig( - ApplicationProtocolConfig.Protocol.ALPN, - ApplicationProtocolConfig.SelectorFailureBehavior.NO_ADVERTISE, - ApplicationProtocolConfig.SelectedListenerFailureBehavior.ACCEPT, - ApplicationProtocolNames.HTTP_2)) - .build() - ) - .build()); - } - - public QueryRouteRequest buildQueryRouteRequest(String topic) { - return buildQueryRouteRequest(topic, Endpoints.getDefaultInstance()); - } - - public QueryRouteRequest buildQueryRouteRequest(String topic, Endpoints endpoints) { - return QueryRouteRequest.newBuilder() - .setTopic(Resource.newBuilder() - .setName(topic) - .build()) - .setEndpoints(endpoints) - .build(); - } - - public SendMessageRequest buildSendMessageRequest(String topic, String messageId) { - return SendMessageRequest.newBuilder() - .setMessage(Message.newBuilder() - .setTopic(Resource.newBuilder() - .setName(topic) - .build()) - .setSystemAttribute(SystemAttribute.newBuilder() - .setMessageId(messageId) - .setPartitionId(0) - .build()) - .setBody(ByteString.copyFromUtf8("123")) - .build()) - .build(); - } - - public SendMessageRequest buildSendDelayMessageRequest(String topic, String messageId, int delayLevel) { -// Message message; -// message.getSystemAttribute().getTimedDeliveryCase(); - return SendMessageRequest.newBuilder() - .setMessage(Message.newBuilder() - .setTopic(Resource.newBuilder() - .setName(topic) - .build()) - .setSystemAttribute(SystemAttribute.newBuilder() - .setMessageId(messageId) - .setPartitionId(0) - .setDelayLevel(delayLevel) - .build()) - .setBody(ByteString.copyFromUtf8("123")) - .build()) - .build(); - } - - public ReceiveMessageRequest buildReceiveMessageRequest(String group, String topic) { - return ReceiveMessageRequest.newBuilder() - .setGroup(Resource.newBuilder() - .setName(group) - .build()) - .setPartition(Partition.newBuilder() - .setTopic(Resource.newBuilder() - .setName(topic) - .build()) - .setId(0) - .build()) - .setBatchSize(16) - .setInvisibleDuration(Duration.newBuilder() - .setSeconds(3) - .build()) - .setInitializationTimestamp(Timestamp.newBuilder() - .setSeconds(TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis())) - .build()) - .build(); - } - - public AckMessageRequest buildAckMessageRequest(String group, String topic, String receiptHandle) { - return AckMessageRequest.newBuilder() - .setGroup(Resource.newBuilder() - .setName(group) - .build()) - .setTopic(Resource.newBuilder() - .setName(topic) - .build()) - .setReceiptHandle(receiptHandle) - .build(); - } - - public void assertQueryRoute(QueryRouteResponse response, int brokerSize) { - assertThat(response.getCommon().getStatus().getCode()).isEqualTo(Code.OK_VALUE); - assertThat(response.getPartitionsList().size()).isEqualTo(brokerSize * defaultQueueNums); - assertThat(response.getPartitions(0).getBroker().getEndpoints().getAddresses(0).getPort()).isEqualTo(ConfigurationManager.getProxyConfig().getGrpcServerPort()); - } - - public void assertSendMessage(SendMessageResponse response, String messageId) { - assertThat(response.getCommon() - .getStatus() - .getCode()).isEqualTo(Code.OK.getNumber()); - assertThat(response.getMessageId()).isEqualTo(messageId); - } - - public void assertReceiveMessage(ReceiveMessageResponse response, String messageId) { - assertThat(response.getCommon() - .getStatus() - .getCode()).isEqualTo(Code.OK.getNumber()); - assertThat(response.getMessagesCount()).isEqualTo(1); - assertThat(response.getMessages(0) - .getSystemAttribute() - .getMessageId()).isEqualTo(messageId); - } - - public void assertAck(AckMessageResponse response) { - assertThat(response.getCommon() - .getStatus() - .getCode()).isEqualTo(Code.OK_VALUE); - } -} \ No newline at end of file diff --git a/test/src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseTest.java b/test/src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseTest.java index 7c4379bacc..b008ead8d0 100644 --- a/test/src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseTest.java +++ b/test/src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseTest.java @@ -500,6 +500,7 @@ public class GrpcBaseTest extends BaseConf { public QueryRouteRequest buildQueryRouteRequest(String topic) { return QueryRouteRequest.newBuilder() + .setEndpoints(buildEndpoints(PORT)) .setTopic(Resource.newBuilder() .setName(topic) .build()) @@ -508,6 +509,7 @@ public class GrpcBaseTest extends BaseConf { public QueryAssignmentRequest buildQueryAssignmentRequest(String topic, String group) { return QueryAssignmentRequest.newBuilder() + .setEndpoints(buildEndpoints(PORT)) .setTopic(Resource.newBuilder().setName(topic).build()) .setGroup(Resource.newBuilder().setName(group).build()) .build(); @@ -683,15 +685,19 @@ public class GrpcBaseTest extends BaseConf { assertThat(response.getReceiptHandle()).isNotEqualTo(prevHandle); } + public Endpoints buildEndpoints(int port) { + return Endpoints.newBuilder() + .setScheme(AddressScheme.IPv4) + .addAddresses(Address.newBuilder() + .setHost("127.0.0.1") + .setPort(port) + .build()) + .build(); + } + public Settings buildAccessPointClientSettings(int port) { return Settings.newBuilder() - .setAccessPoint(Endpoints.newBuilder() - .setScheme(AddressScheme.IPv4) - .addAddresses(Address.newBuilder() - .setHost("127.0.0.1") - .setPort(port) - .build()) - .build()) + .setAccessPoint(buildEndpoints(port)) .build(); }