diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/HealthCheckServer.java b/proxy/src/main/java/org/apache/rocketmq/proxy/HealthCheckServer.java index ed5e977a39..789a0dcf29 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/HealthCheckServer.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/HealthCheckServer.java @@ -23,6 +23,7 @@ import com.sun.net.httpserver.HttpServer; import java.io.IOException; import java.io.OutputStream; import java.net.InetSocketAddress; +import java.util.concurrent.TimeUnit; import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.proxy.common.StartAndShutdown; @@ -39,8 +40,9 @@ public class HealthCheckServer implements StartAndShutdown { } @Override - public void shutdown() { + public void shutdown() throws InterruptedException { this.healthChecker.stop(0); + Thread.sleep(TimeUnit.SECONDS.toMillis(ConfigurationManager.getProxyConfig().getWaitAfterStopHealthCheckInSeconds())); } static class HealthCheckHandler implements HttpHandler { diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java b/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java index 808d748b7b..810fd8c1ed 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java @@ -21,11 +21,12 @@ import ch.qos.logback.classic.LoggerContext; import ch.qos.logback.classic.joran.JoranConfigurator; import ch.qos.logback.core.joran.spi.JoranException; import java.util.Date; -import java.util.concurrent.TimeUnit; import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.broker.BrokerStartup; import org.apache.rocketmq.client.log.ClientLogger; import org.apache.rocketmq.common.thread.ThreadPoolMonitor; +import org.apache.rocketmq.proxy.common.AbstractStartAndShutdown; +import org.apache.rocketmq.proxy.common.StartAndShutdown; import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.proxy.config.ProxyConfig; import org.apache.rocketmq.proxy.grpc.GrpcServer; @@ -39,6 +40,13 @@ import org.slf4j.LoggerFactory; public class ProxyStartup { private static final Logger log = LoggerFactory.getLogger(ProxyStartup.class); + private static final ProxyStartAndShutdown proxyStartAndShutdown = new ProxyStartAndShutdown(); + + private static class ProxyStartAndShutdown extends AbstractStartAndShutdown { + @Override public void appendStartAndShutdown(StartAndShutdown startAndShutdown) { + super.appendStartAndShutdown(startAndShutdown); + } + } public static void main(String[] args) { try { @@ -51,26 +59,19 @@ public class ProxyStartup { // create and start grpcServer GrpcServer grpcServer = createGrpcServer(); - grpcServer.start(); + proxyStartAndShutdown.appendStartAndShutdown(grpcServer); // health check server final HealthCheckServer healthCheckServer = new HealthCheckServer(); - healthCheckServer.start(); + proxyStartAndShutdown.appendStartAndShutdown(healthCheckServer); Runtime.getRuntime().addShutdownHook(new Thread(() -> { log.info("try to shutdown server"); try { - healthCheckServer.shutdown(); - Thread.sleep(TimeUnit.SECONDS.toMillis(ConfigurationManager.getProxyConfig().getWaitAfterStopHealthCheckInSeconds())); + proxyStartAndShutdown.shutdown(); } catch (Exception e) { - log.error("err when shutdown healthCheckServer", e); - } - - try { - grpcServer.shutdown(); - } catch (Exception e) { - log.error("err when shutdown grpc server", e); + log.error("err when shutdown proxy", e); } })); } catch (Exception e) { @@ -84,20 +85,29 @@ public class ProxyStartup { log.info(new Date() + "rmq-proxy startup successfully"); } - private static GrpcServer createGrpcServer() throws RuntimeException { + private static GrpcServer createGrpcServer() throws Exception { GrpcForwardService grpcService; String proxyModeStr = ConfigurationManager.getProxyConfig().getProxyMode(); if (ProxyMode.isClusterMode(proxyModeStr)) { grpcService = new ClusterGrpcService(); } else if (ProxyMode.isLocalMode(proxyModeStr)) { BrokerController brokerController = createBrokerController(); + StartAndShutdown brokerControllerWrapper = new StartAndShutdown() { + @Override public void start() throws Exception { + brokerController.start(); + } + + @Override public void shutdown() throws Exception { + brokerController.shutdown(); + } + }; + proxyStartAndShutdown.appendStartAndShutdown(brokerControllerWrapper); grpcService = new LocalGrpcService(brokerController); } else { throw new IllegalArgumentException("try to start grpc server with wrong mode, use 'local' or 'cluster'"); } return new GrpcServer(grpcService); - } private static BrokerController createBrokerController() { diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServer.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServer.java index b2b0dafb38..c5e318a245 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServer.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServer.java @@ -31,6 +31,7 @@ import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.common.thread.ThreadPoolMonitor; +import org.apache.rocketmq.proxy.common.StartAndShutdown; import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.proxy.grpc.interceptor.ContextInterceptor; import org.apache.rocketmq.proxy.grpc.interceptor.HeaderInterceptor; @@ -38,7 +39,7 @@ import org.apache.rocketmq.proxy.grpc.service.GrpcForwardService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public class GrpcServer { +public class GrpcServer implements StartAndShutdown { private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.GRPC_LOGGER_NAME); private final io.grpc.Server server; diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java index 89be32779e..ce4e7f78b1 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java @@ -86,23 +86,25 @@ import org.apache.rocketmq.common.protocol.heartbeat.HeartbeatData; import org.apache.rocketmq.proxy.channel.ChannelManager; import org.apache.rocketmq.proxy.channel.SimpleChannel; import org.apache.rocketmq.proxy.channel.SimpleChannelHandlerContext; +import org.apache.rocketmq.proxy.common.AbstractStartAndShutdown; +import org.apache.rocketmq.proxy.common.StartAndShutdown; import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.proxy.connector.ConnectorManager; import org.apache.rocketmq.proxy.grpc.adapter.InvocationContext; import org.apache.rocketmq.proxy.grpc.adapter.channel.GrpcClientChannel; +import org.apache.rocketmq.proxy.grpc.adapter.channel.PullMessageChannel; import org.apache.rocketmq.proxy.grpc.adapter.channel.ReceiveMessageChannel; import org.apache.rocketmq.proxy.grpc.adapter.channel.SendMessageChannel; -import org.apache.rocketmq.proxy.grpc.adapter.channel.PullMessageChannel; import org.apache.rocketmq.proxy.grpc.adapter.handler.PullMessageResponseHandler; import org.apache.rocketmq.proxy.grpc.adapter.handler.ReceiveMessageResponseHandler; import org.apache.rocketmq.proxy.grpc.adapter.handler.SendMessageResponseHandler; import org.apache.rocketmq.proxy.grpc.common.Converter; import org.apache.rocketmq.proxy.grpc.common.DelayPolicy; -import org.apache.rocketmq.proxy.grpc.interceptor.InterceptorConstants; import org.apache.rocketmq.proxy.grpc.common.PollCommandResponseFuture; import org.apache.rocketmq.proxy.grpc.common.PollCommandResponseManager; import org.apache.rocketmq.proxy.grpc.common.ProxyMode; import org.apache.rocketmq.proxy.grpc.common.ResponseBuilder; +import org.apache.rocketmq.proxy.grpc.interceptor.InterceptorConstants; import org.apache.rocketmq.proxy.grpc.service.cluster.RouteService; import org.apache.rocketmq.remoting.RemotingServer; import org.apache.rocketmq.remoting.netty.NettyRemotingAbstract; @@ -111,7 +113,7 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public class LocalGrpcService implements GrpcForwardService { +public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcForwardService { private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.GRPC_LOGGER_NAME); private final BrokerController brokerController; @@ -130,6 +132,8 @@ public class LocalGrpcService implements GrpcForwardService { this.pollCommandResponseManager = new PollCommandResponseManager(); this.routeService = new RouteService(ProxyMode.LOCAL, connectorManager); this.delayPolicy = DelayPolicy.build(brokerController.getMessageStoreConfig().getMessageDelayLevel()); + this.appendStartAndShutdown(connectorManager); + this.appendStartAndShutdown(new LocalGrpcServiceStartAndShutdown()); } @Override public CompletableFuture queryRoute(Context ctx, QueryRouteRequest request) { @@ -547,14 +551,14 @@ public class LocalGrpcService implements GrpcForwardService { return future; } - @Override public void start() throws Exception { - this.brokerController.start(); - this.scheduledExecutorService.scheduleWithFixedDelay(this::scanAndCleanChannels, 5, 5, TimeUnit.MINUTES); - } + private class LocalGrpcServiceStartAndShutdown implements StartAndShutdown { + @Override public void start() throws Exception { + LocalGrpcService.this.scheduledExecutorService.scheduleWithFixedDelay(LocalGrpcService.this::scanAndCleanChannels, 5, 5, TimeUnit.MINUTES); + } - @Override public void shutdown() throws Exception { - this.scheduledExecutorService.shutdown(); - this.brokerController.shutdown(); + @Override public void shutdown() throws Exception { + LocalGrpcService.this.scheduledExecutorService.shutdown(); + } } private void scanAndCleanChannels() { diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/RouteService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/RouteService.java index e7c64036b5..37d4acb047 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/RouteService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/RouteService.java @@ -41,6 +41,7 @@ import org.apache.rocketmq.common.constant.PermName; import org.apache.rocketmq.common.protocol.route.BrokerData; import org.apache.rocketmq.common.protocol.route.QueueData; import org.apache.rocketmq.common.protocol.route.TopicRouteData; +import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.proxy.connector.ConnectorManager; import org.apache.rocketmq.proxy.connector.route.MessageQueueWrapper; import org.apache.rocketmq.proxy.connector.route.SelectableMessageQueue; @@ -293,7 +294,7 @@ public class RouteService extends BaseService { .setScheme(AddressScheme.IPv4) .addAddresses( Address.newBuilder() - .setPort(hostAndPort.getPort()) + .setPort(ConfigurationManager.getProxyConfig().getGrpcServerPort()) .setHost(hostAndPort.getHost()) ) .build()) diff --git a/test/src/test/java/org/apache/rocketmq/test/base/GrpcBaseTest.java b/test/src/test/java/org/apache/rocketmq/test/base/GrpcBaseTest.java index 25729bf16e..271dc0b75c 100644 --- a/test/src/test/java/org/apache/rocketmq/test/base/GrpcBaseTest.java +++ b/test/src/test/java/org/apache/rocketmq/test/base/GrpcBaseTest.java @@ -17,6 +17,20 @@ package org.apache.rocketmq.test.base; +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; @@ -31,13 +45,14 @@ 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.junit.Rule; -import org.junit.runner.RunWith; -import org.mockito.junit.MockitoJUnitRunner; -@RunWith(MockitoJUnitRunner.class) +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. @@ -45,6 +60,8 @@ public class GrpcBaseTest extends BaseConf { @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(); @@ -75,4 +92,71 @@ public class GrpcBaseTest extends BaseConf { ) .build()); } + + public QueryRouteRequest buildQueryRouteRequest(String topic) { + return QueryRouteRequest.newBuilder() + .setTopic(Resource.newBuilder() + .setName(topic) + .build()) + .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 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 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); + } } \ No newline at end of file diff --git a/test/src/test/java/org/apache/rocketmq/test/proxy/LocalGrpcTest.java b/test/src/test/java/org/apache/rocketmq/test/proxy/LocalGrpcTest.java index 387764ad3b..16e3d54326 100644 --- a/test/src/test/java/org/apache/rocketmq/test/proxy/LocalGrpcTest.java +++ b/test/src/test/java/org/apache/rocketmq/test/proxy/LocalGrpcTest.java @@ -17,19 +17,10 @@ package org.apache.rocketmq.test.proxy; -import apache.rocketmq.v1.Message; import apache.rocketmq.v1.MessagingServiceGrpc; -import apache.rocketmq.v1.Partition; -import apache.rocketmq.v1.ReceiveMessageRequest; +import apache.rocketmq.v1.QueryRouteResponse; 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 java.net.URL; import java.util.concurrent.TimeUnit; @@ -37,15 +28,16 @@ import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.proxy.grpc.GrpcMessagingProcessor; import org.apache.rocketmq.proxy.grpc.service.LocalGrpcService; import org.apache.rocketmq.test.base.GrpcBaseTest; +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.assertj.core.api.Assertions.assertThat; public class LocalGrpcTest extends GrpcBaseTest { private MessagingServiceGrpc.MessagingServiceBlockingStub blockingStub; + private LocalGrpcService localGrpcService; @Before public void setUp() throws Exception { @@ -57,56 +49,37 @@ public class LocalGrpcTest extends GrpcBaseTest { System.setProperty(RMQ_PROXY_HOME, mockProxyHome); ConfigurationManager.initEnv(); ConfigurationManager.intConfig(); - GrpcMessagingProcessor processor = new GrpcMessagingProcessor(new LocalGrpcService(brokerController1)); ConfigurationManager.getProxyConfig().setGrpcServerPort(8082); + ConfigurationManager.getProxyConfig().setNameSrvAddr(nsAddr); + localGrpcService = new LocalGrpcService(brokerController1); + localGrpcService.start(); + GrpcMessagingProcessor processor = new GrpcMessagingProcessor(localGrpcService); Channel channel = setUpServer(processor, ConfigurationManager.getProxyConfig().getGrpcServerPort(), true); blockingStub = MessagingServiceGrpc.newBlockingStub(channel); } + @After + public void clean() throws Exception { + localGrpcService.shutdown(); + shutdown(); + } + + @Test + public void testQueryRoute() { + String topic = initTopic(); + QueryRouteResponse response = blockingStub.queryRoute(buildQueryRouteRequest(topic)); + assertQueryRoute(response, brokerControllerList.size()); + } + @Test public void testSendReceiveMessage() { String group = "group"; - SendMessageResponse sendResponse = blockingStub.sendMessage(SendMessageRequest.newBuilder() - .setMessage(Message.newBuilder() - .setTopic(Resource.newBuilder() - .setName(broker1Name) - .build()) - .setSystemAttribute(SystemAttribute.newBuilder() - .setMessageId(createUniqID()) - .setPartitionId(0) - .build()) - .setBody(ByteString.copyFromUtf8("123")) - .build()) - .build()); - assertThat(sendResponse.getCommon() - .getStatus() - .getCode()).isEqualTo(Code.OK.getNumber()); - String messageId = sendResponse.getMessageId(); + String messageId = createUniqID(); + SendMessageResponse sendResponse = blockingStub.sendMessage(buildSendMessageRequest(broker1Name, messageId)); + assertSendMessage(sendResponse, messageId); + ReceiveMessageResponse receiveResponse = blockingStub.withDeadlineAfter(3, TimeUnit.SECONDS) - .receiveMessage(ReceiveMessageRequest.newBuilder() - .setGroup(Resource.newBuilder() - .setName(group) - .build()) - .setPartition(Partition.newBuilder() - .setTopic(Resource.newBuilder() - .setName(broker1Name) - .build()) - .setId(0) - .build()) - .setBatchSize(16) - .setInvisibleDuration(Duration.newBuilder() - .setSeconds(3) - .build()) - .setInitializationTimestamp(Timestamp.newBuilder() - .setSeconds(TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis())) - .build()) - .build()); - assertThat(receiveResponse.getCommon() - .getStatus() - .getCode()).isEqualTo(Code.OK.getNumber()); - assertThat(receiveResponse.getMessagesCount()).isEqualTo(1); - assertThat(receiveResponse.getMessages(0) - .getSystemAttribute() - .getMessageId()).isEqualTo(messageId); + .receiveMessage(buildReceiveMessageRequest(group, broker1Name)); + assertReceiveMessage(receiveResponse, messageId); } }