opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-24 16:04:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #3949] Refector GrpcBaseTest

Browse Source
This commit is contained in:
zhouxiang
2022-07-13 11:29:13 +08:00
parent 3036cf421d
commit 2824c686db
7 changed files with 158 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/HealthCheckServer.java
+3 -1
View File
@@ -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 {
proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java
+24 -14
View File
@@ -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() {
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServer.java
+2 -1
View File
@@ -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;
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java
+14 -10
View File
@@ -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<QueryRouteResponse> 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() {
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/RouteService.java
+2 -1
View File
@@ -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())
test/src/test/java/org/apache/rocketmq/test/base/GrpcBaseTest.java
+87 -3
View File
@@ -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);
}
}
test/src/test/java/org/apache/rocketmq/test/proxy/LocalGrpcTest.java
+26 -53
View File
@@ -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);
}
}