From a45b040fba5f051a81ae2ddb0b9d0e2fde86cb17 Mon Sep 17 00:00:00 2001 From: "kaiyi.lk" Date: Mon, 23 May 2022 19:12:58 +0800 Subject: [PATCH] [ISSUE #3949] add test cases --- .../proxy/grpc/v2/client/ClientActivity.java | 16 +- .../proxy/service/ClusterServiceManager.java | 14 +- .../metadata/ClusterMetadataService.java | 18 +- .../service/route/MessageQueueSelector.java | 26 -- .../service/route/TopicRouteService.java | 8 +- .../ClusterTransactionService.java | 27 +- .../proxy/grpc/v2/BaseActivityTest.java | 8 +- .../grpc/v2/client/ClientActivityTest.java | 259 ++++++++++++++++++ .../ForwardMessageToDLQActivityTest.java | 64 +++++ .../EndTransactionActivityTest.java | 101 +++++++ .../proxy/service/BaseServiceTest.java | 83 ++++++ .../metadata/ClusterMetadataServiceTest.java | 73 +++++ .../route/ClusterTopicRouteServiceTest.java | 69 +++++ .../route/LocalTopicRouteServiceTest.java | 102 +++++++ .../route/MessageQueueSelectorTest.java | 84 ++++++ .../ClusterTransactionServiceTest.java | 141 ++++++++++ 16 files changed, 1028 insertions(+), 65 deletions(-) create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/ForwardMessageToDLQActivityTest.java create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/transaction/EndTransactionActivityTest.java create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/service/BaseServiceTest.java create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/service/metadata/ClusterMetadataServiceTest.java create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/service/route/ClusterTopicRouteServiceTest.java create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/service/route/LocalTopicRouteServiceTest.java create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelectorTest.java create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionServiceTest.java diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java index 5cbcc94244..e4a4e60918 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java @@ -129,7 +129,10 @@ public class ClientActivity extends AbstractMessingActivity { break; } default: { - throw new IllegalArgumentException("ClientType not exist " + clientSettings.getClientType()); + future.complete(HeartbeatResponse.newBuilder() + .setStatus(ResponseBuilder.buildStatus(Code.UNRECOGNIZED_CLIENT_TYPE, clientSettings.getClientType().name())) + .build()); + return future; } } future.complete(HeartbeatResponse.newBuilder() @@ -150,7 +153,7 @@ public class ClientActivity extends AbstractMessingActivity { ProxyContext context = createContext(ctx); String clientId = context.getVal(GrpcContextConstants.CLIENT_ID); LanguageCode languageCode = context.getVal(GrpcContextConstants.LANGUAGE); - Settings clientSettings = grpcClientSettingsManager.getClientSettings(context); + Settings clientSettings = grpcClientSettingsManager.removeClientSettings(clientId); switch (clientSettings.getClientType()) { case PRODUCER: @@ -158,8 +161,8 @@ public class ClientActivity extends AbstractMessingActivity { String topicName = GrpcConverter.wrapResourceWithNamespace(topic); // user topic name as producer group GrpcClientChannel channel = this.grpcChannelManager.removeChannel(topicName, clientId); - ClientChannelInfo clientChannelInfo = new ClientChannelInfo(channel, clientId, languageCode, MQVersion.Version.V5_0_0.ordinal()); if (channel != null) { + ClientChannelInfo clientChannelInfo = new ClientChannelInfo(channel, clientId, languageCode, MQVersion.Version.V5_0_0.ordinal()); this.messagingProcessor.unRegisterProducer(context, topicName, clientChannelInfo); } } @@ -171,13 +174,16 @@ public class ClientActivity extends AbstractMessingActivity { } String consumerGroup = GrpcConverter.wrapResourceWithNamespace(request.getGroup()); GrpcClientChannel channel = this.grpcChannelManager.removeChannel(consumerGroup, clientId); - ClientChannelInfo clientChannelInfo = new ClientChannelInfo(channel, clientId, languageCode, MQVersion.Version.V5_0_0.ordinal()); if (channel != null) { + ClientChannelInfo clientChannelInfo = new ClientChannelInfo(channel, clientId, languageCode, MQVersion.Version.V5_0_0.ordinal()); this.messagingProcessor.unRegisterConsumer(context, consumerGroup, clientChannelInfo); } break; default: - break; + future.complete(NotifyClientTerminationResponse.newBuilder() + .setStatus(ResponseBuilder.buildStatus(Code.UNRECOGNIZED_CLIENT_TYPE, clientSettings.getClientType().name())) + .build()); + return future; } future.complete(NotifyClientTerminationResponse.newBuilder() .setStatus(ResponseBuilder.buildStatus(Code.OK, Code.OK.name())) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/ClusterServiceManager.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/ClusterServiceManager.java index 91604cb13b..ce529b16e1 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/ClusterServiceManager.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/ClusterServiceManager.java @@ -37,6 +37,7 @@ import org.apache.rocketmq.proxy.service.message.MessageService; import org.apache.rocketmq.proxy.service.metadata.ClusterMetadataService; import org.apache.rocketmq.proxy.service.metadata.MetadataService; import org.apache.rocketmq.proxy.service.mqclient.DoNothingClientRemotingProcessor; +import org.apache.rocketmq.proxy.service.mqclient.ProxyClientRemotingProcessor; import org.apache.rocketmq.proxy.service.relay.ClusterProxyRelayService; import org.apache.rocketmq.proxy.service.relay.ProxyRelayService; import org.apache.rocketmq.proxy.service.mqclient.MQClientAPIFactory; @@ -60,6 +61,7 @@ public class ClusterServiceManager extends AbstractStartAndShutdown implements S private final ScheduledExecutorService scheduledExecutorService; private final MQClientAPIFactory messagingClientAPIFactory; private final MQClientAPIFactory operationClientAPIFactory; + private final MQClientAPIFactory transactionClientAPIFactory; public ClusterServiceManager(RPCHook rpcHook) { this.scheduledExecutorService = Executors.newScheduledThreadPool(3); @@ -68,7 +70,7 @@ public class ClusterServiceManager extends AbstractStartAndShutdown implements S ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig(); this.messagingClientAPIFactory = new MQClientAPIFactory( - "CLUSTER_MQ_CLIENT_", + "ClusterMQClient_", proxyConfig.getRocketmqMQClientNum(), new DoNothingClientRemotingProcessor(null), rpcHook, @@ -80,10 +82,17 @@ public class ClusterServiceManager extends AbstractStartAndShutdown implements S rpcHook, this.scheduledExecutorService ); + this.transactionClientAPIFactory = new MQClientAPIFactory( + "ClusterTransaction_", + 1, + new ProxyClientRemotingProcessor(producerManager), + rpcHook, + scheduledExecutorService); this.topicRouteService = new ClusterTopicRouteService(operationClientAPIFactory); this.messageService = new ClusterMessageService(this.topicRouteService, this.messagingClientAPIFactory); - this.clusterTransactionService = new ClusterTransactionService(this.topicRouteService, this.producerManager, rpcHook); + this.clusterTransactionService = new ClusterTransactionService(this.topicRouteService, this.producerManager, rpcHook, + this.transactionClientAPIFactory); this.proxyRelayService = new ClusterProxyRelayService(); this.metadataService = new ClusterMetadataService(topicRouteService, operationClientAPIFactory); @@ -105,6 +114,7 @@ public class ClusterServiceManager extends AbstractStartAndShutdown implements S this.appendShutdown(scheduledExecutorService::shutdown); this.appendStartAndShutdown(this.messagingClientAPIFactory); this.appendStartAndShutdown(this.operationClientAPIFactory); + this.appendStartAndShutdown(this.transactionClientAPIFactory); this.appendStartAndShutdown(this.topicRouteService); this.appendStartAndShutdown(this.clusterTransactionService); this.appendStartAndShutdown(this.metadataService); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/metadata/ClusterMetadataService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/metadata/ClusterMetadataService.java index a5b18636d6..0291289340 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/metadata/ClusterMetadataService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/metadata/ClusterMetadataService.java @@ -22,7 +22,6 @@ import com.google.common.cache.LoadingCache; import java.util.Optional; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; -import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.common.attribute.TopicMessageType; import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.common.protocol.route.BrokerData; @@ -43,15 +42,16 @@ public class ClusterMetadataService extends AbstractStartAndShutdown implements protected static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); private static final long DEFAULT_TIMEOUT = 3000; - private final ThreadPoolExecutor cacheRefreshExecutor; private final TopicRouteService topicRouteService; private final MQClientAPIFactory mqClientAPIFactory; - private final LoadingCache topicCache; - private final static TopicConfigAndQueueMapping EMPTY_TOPIC_CONFIG = new TopicConfigAndQueueMapping(); + protected final ThreadPoolExecutor cacheRefreshExecutor; - private final LoadingCache subscriptionGroupConfigCache; - private final static SubscriptionGroupConfig EMPTY_SUBSCRIPTION_GROUP_CONFIG = new SubscriptionGroupConfig(); + protected final LoadingCache topicConfigCache; + protected final static TopicConfigAndQueueMapping EMPTY_TOPIC_CONFIG = new TopicConfigAndQueueMapping(); + + protected final LoadingCache subscriptionGroupConfigCache; + protected final static SubscriptionGroupConfig EMPTY_SUBSCRIPTION_GROUP_CONFIG = new SubscriptionGroupConfig(); public ClusterMetadataService(TopicRouteService topicRouteService, MQClientAPIFactory mqClientAPIFactory) { this.topicRouteService = topicRouteService; @@ -66,7 +66,7 @@ public class ClusterMetadataService extends AbstractStartAndShutdown implements "MetadataCacheRefresh", config.getMetadataThreadPoolQueueCapacity() ); - this.topicCache = CacheBuilder.newBuilder() + this.topicConfigCache = CacheBuilder.newBuilder() .maximumSize(config.getTopicConfigCacheMaxNum()) .refreshAfterWrite(config.getTopicConfigCacheExpiredInSeconds(), TimeUnit.SECONDS) .build(new ClusterTopicConfigCacheLoader()); @@ -86,7 +86,7 @@ public class ClusterMetadataService extends AbstractStartAndShutdown implements public TopicMessageType getTopicMessageType(String topic) { TopicConfigAndQueueMapping topicConfigAndQueueMapping; try { - topicConfigAndQueueMapping = topicCache.get(topic); + topicConfigAndQueueMapping = topicConfigCache.get(topic); } catch (Exception e) { return TopicMessageType.UNSPECIFIED; } @@ -159,7 +159,7 @@ public class ClusterMetadataService extends AbstractStartAndShutdown implements protected Optional findOneBroker(String topic) throws Exception { try { return topicRouteService.getAllMessageQueueView(topic).getTopicRouteData().getBrokerDatas().stream().findAny(); - } catch (MQClientException e) { + } catch (Exception e) { if (TopicRouteHelper.isTopicNotExistError(e)) { return Optional.empty(); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelector.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelector.java index f37155e319..4c4ccdc49a 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelector.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelector.java @@ -152,16 +152,6 @@ public class MessageQueueSelector { return selectOneByIndex(nextIndex, onlyBroker); } - public final SelectableMessageQueue selectOne(String brokerName, int queueId) { - for (SelectableMessageQueue targetMessageQueue : queues) { - String queueBrokerName = targetMessageQueue.getBrokerName(); - if (queueBrokerName.equals(brokerName) && targetMessageQueue.getQueueId() == queueId) { - return targetMessageQueue; - } - } - return null; - } - public final SelectableMessageQueue selectOneByIndex(int index, boolean onlyBroker) { if (onlyBroker) { if (brokerActingQueues.isEmpty()) { @@ -176,22 +166,6 @@ public class MessageQueueSelector { return queues.get(Math.abs(index) % queues.size()); } - // find next same type(but different) queue with last(normal queue or broker acting queue). - public final SelectableMessageQueue selectNextQueue(SelectableMessageQueue last) { - boolean onlyBroker = last.getQueueId() < 0; - SelectableMessageQueue newOne = last; - int count = onlyBroker ? brokerActingQueues.size() : queues.size(); - - for (int i = 0; i < count; i++) { - newOne = selectOne(onlyBroker); - if (!newOne.getBrokerName().equals(last.getBrokerName()) || newOne.getQueueId() != last.getQueueId()) { - break; - } - } - - return newOne; - } - public List getQueues() { return queues; } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/TopicRouteService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/TopicRouteService.java index c800abfa1a..fc9edad5c8 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/TopicRouteService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/TopicRouteService.java @@ -42,11 +42,11 @@ import org.apache.rocketmq.proxy.service.mqclient.MQClientAPIFactory; public abstract class TopicRouteService extends AbstractStartAndShutdown { private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); - private final LoadingCache topicCache; - private final MQClientAPIFactory mqClientAPIFactory; - private final ScheduledExecutorService scheduledExecutorService; - private final ThreadPoolExecutor cacheRefreshExecutor; + + protected final LoadingCache topicCache; + protected final ScheduledExecutorService scheduledExecutorService; + protected final ThreadPoolExecutor cacheRefreshExecutor; public TopicRouteService(MQClientAPIFactory mqClientAPIFactory) { ProxyConfig config = ConfigurationManager.getProxyConfig(); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionService.java index c892d3ef74..210f092203 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionService.java @@ -26,13 +26,10 @@ import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import org.apache.rocketmq.broker.client.ProducerManager; import org.apache.rocketmq.common.ServiceThread; -import org.apache.rocketmq.common.ThreadFactoryImpl; import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.common.protocol.heartbeat.HeartbeatData; import org.apache.rocketmq.common.protocol.heartbeat.ProducerData; @@ -44,7 +41,6 @@ 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.service.mqclient.MQClientAPIFactory; -import org.apache.rocketmq.proxy.service.mqclient.ProxyClientRemotingProcessor; import org.apache.rocketmq.proxy.service.route.MessageQueueView; import org.apache.rocketmq.proxy.service.route.TopicRouteService; import org.apache.rocketmq.remoting.RPCHook; @@ -57,21 +53,14 @@ public class ClusterTransactionService implements StartAndShutdown, TransactionS private final MQClientAPIFactory mqClientAPIFactory; private final TopicRouteService topicRouteService; - private final ScheduledExecutorService scheduledExecutorService; private ThreadPoolExecutor heartbeatExecutors; private final Map/* cluster list */> groupClusterData = new ConcurrentHashMap<>(); private TxHeartbeatServiceThread txHeartbeatServiceThread; - public ClusterTransactionService(TopicRouteService topicRouteService, ProducerManager producerManager, RPCHook rpcHook) { - this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor( - new ThreadFactoryImpl("ClusterTransactionScheduledThread_")); + public ClusterTransactionService(TopicRouteService topicRouteService, ProducerManager producerManager, RPCHook rpcHook, + MQClientAPIFactory mqClientAPIFactory) { this.topicRouteService = topicRouteService; - this.mqClientAPIFactory = new MQClientAPIFactory( - "ClusterTransaction_", - 1, - new ProxyClientRemotingProcessor(producerManager), - rpcHook, - scheduledExecutorService); + this.mqClientAPIFactory = mqClientAPIFactory; } @Override @@ -182,6 +171,10 @@ public class ClusterTransactionService implements StartAndShutdown, TransactionS } } + public Map> getGroupClusterData() { + return groupClusterData; + } + protected void sendHeartBeatToCluster(String clusterName, List heartbeatDataList) { if (heartbeatDataList == null) { return; @@ -220,6 +213,10 @@ public class ClusterTransactionService implements StartAndShutdown, TransactionS this.cluster = cluster; } + public String getCluster() { + return cluster; + } + @Override public boolean equals(Object obj) { if (obj == this) { @@ -264,7 +261,6 @@ public class ClusterTransactionService implements StartAndShutdown, TransactionS ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig(); txHeartbeatServiceThread = new TxHeartbeatServiceThread(); - mqClientAPIFactory.start(); txHeartbeatServiceThread.start(); heartbeatExecutors = ThreadPoolMonitor.createAndMonitor( proxyConfig.getTransactionHeartbeatThreadPoolNums(), @@ -279,6 +275,5 @@ public class ClusterTransactionService implements StartAndShutdown, TransactionS public void shutdown() throws Exception { txHeartbeatServiceThread.shutdown(); heartbeatExecutors.shutdown(); - mqClientAPIFactory.shutdown(); } } diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/BaseActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/BaseActivityTest.java index 751c62f7c0..c17cca7c2c 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/BaseActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/BaseActivityTest.java @@ -34,17 +34,15 @@ import org.junit.runner.RunWith; import org.mockito.Mock; import org.mockito.junit.MockitoJUnitRunner; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @Ignore @RunWith(MockitoJUnitRunner.Silent.class) public class BaseActivityTest extends InitConfigAndLoggerTest { protected static final Random RANDOM = new Random(); - @Mock protected MessagingProcessor messagingProcessor; - @Mock protected GrpcClientSettingsManager grpcClientSettingsManager; - @Mock protected ProxyRelayService proxyRelayService; protected static final String REMOTE_ADDR = "192.168.0.1:8080"; @@ -55,6 +53,10 @@ public class BaseActivityTest extends InitConfigAndLoggerTest { public void before() throws Throwable { super.before(); + messagingProcessor = mock(MessagingProcessor.class); + grpcClientSettingsManager = mock(GrpcClientSettingsManager.class); + proxyRelayService = mock(ProxyRelayService.class); + metadata.put(InterceptorConstants.CLIENT_ID, CLIENT_ID); metadata.put(InterceptorConstants.LANGUAGE, "JAVA"); metadata.put(InterceptorConstants.REMOTE_ADDRESS, REMOTE_ADDR); diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java new file mode 100644 index 0000000000..4accbd2e5f --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java @@ -0,0 +1,259 @@ +/* + * 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.proxy.grpc.v2.client; + +import apache.rocketmq.v2.ClientType; +import apache.rocketmq.v2.Code; +import apache.rocketmq.v2.FilterExpression; +import apache.rocketmq.v2.FilterType; +import apache.rocketmq.v2.HeartbeatRequest; +import apache.rocketmq.v2.HeartbeatResponse; +import apache.rocketmq.v2.NotifyClientTerminationRequest; +import apache.rocketmq.v2.NotifyClientTerminationResponse; +import apache.rocketmq.v2.Publishing; +import apache.rocketmq.v2.Resource; +import apache.rocketmq.v2.Settings; +import apache.rocketmq.v2.Subscription; +import apache.rocketmq.v2.SubscriptionEntry; +import apache.rocketmq.v2.TelemetryCommand; +import io.grpc.Context; +import io.grpc.stub.StreamObserver; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import org.apache.rocketmq.broker.client.ClientChannelInfo; +import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData; +import org.apache.rocketmq.proxy.grpc.v2.BaseActivityTest; +import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcClientChannel; +import org.apache.rocketmq.remoting.protocol.LanguageCode; +import org.assertj.core.util.Lists; +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentCaptor; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.when; + +public class ClientActivityTest extends BaseActivityTest { + + private static final String TOPIC = "topic"; + private static final String CONSUMER_GROUP = "consumerGroup"; + + private ClientActivity clientActivity; + + @Before + public void before() throws Throwable { + super.before(); + this.clientActivity = new ClientActivity(this.messagingProcessor, this.grpcClientSettingsManager); + } + + protected TelemetryCommand sendProducerTelemetry(Context context) throws Throwable { + return this.sendClientTelemetry( + context, + Settings.newBuilder() + .setClientType(ClientType.PRODUCER) + .setPublishing(Publishing.newBuilder() + .addTopics(Resource.newBuilder().setName(TOPIC).build()) + .build()) + .build()).get(); + } + + protected HeartbeatResponse sendProducerHeartbeat(Context context) throws Throwable { + return this.clientActivity.heartbeat(context, HeartbeatRequest.newBuilder() + .setClientType(ClientType.PRODUCER) + .build()).get(); + } + + @Test + public void testProducerHeartbeat() throws Throwable { + Context context = createContext(); + + this.sendProducerTelemetry(context); + + ArgumentCaptor registerProducerGroupArgumentCaptor = ArgumentCaptor.forClass(String.class); + ArgumentCaptor channelInfoArgumentCaptor = ArgumentCaptor.forClass(ClientChannelInfo.class); + doNothing().when(this.messagingProcessor).registerProducer(any(), + registerProducerGroupArgumentCaptor.capture(), + channelInfoArgumentCaptor.capture()); + + ArgumentCaptor txProducerGroupArgumentCaptor = ArgumentCaptor.forClass(String.class); + ArgumentCaptor txProducerTopicArgumentCaptor = ArgumentCaptor.forClass(String.class); + doNothing().when(this.messagingProcessor).addTransactionSubscription(any(), + txProducerGroupArgumentCaptor.capture(), + txProducerTopicArgumentCaptor.capture() + ); + + HeartbeatResponse response = this.sendProducerHeartbeat(context); + + assertEquals(Code.OK, response.getStatus().getCode()); + + assertEquals(Lists.newArrayList(TOPIC), registerProducerGroupArgumentCaptor.getAllValues()); + ClientChannelInfo clientChannelInfo = channelInfoArgumentCaptor.getValue(); + assertClientChannelInfo(clientChannelInfo, TOPIC); + + assertEquals(Lists.newArrayList(TOPIC), txProducerGroupArgumentCaptor.getAllValues()); + assertEquals(Lists.newArrayList(TOPIC), txProducerTopicArgumentCaptor.getAllValues()); + } + + protected TelemetryCommand sendConsumerTelemetry(Context context) throws Throwable { + return this.sendClientTelemetry( + context, + Settings.newBuilder() + .setClientType(ClientType.PUSH_CONSUMER) + .setSubscription(Subscription.newBuilder() + .addSubscriptions(SubscriptionEntry.newBuilder() + .setExpression(FilterExpression.newBuilder() + .setExpression("tag") + .setType(FilterType.TAG) + .build()) + .setTopic(Resource.newBuilder().setName(TOPIC).build()) + .build()) + .build()) + .build()).get(); + } + + protected HeartbeatResponse sendConsumerHeartbeat(Context context) throws Throwable { + return this.clientActivity.heartbeat(context, HeartbeatRequest.newBuilder() + .setClientType(ClientType.PUSH_CONSUMER) + .setGroup(Resource.newBuilder().setName(CONSUMER_GROUP).build()) + .build()).get(); + } + + @Test + public void testConsumerHeartbeat() throws Throwable { + Context context = createContext(); + this.sendConsumerTelemetry(context); + + ArgumentCaptor> subscriptionDatasArgumentCaptor = ArgumentCaptor.forClass(Set.class); + ArgumentCaptor channelInfoArgumentCaptor = ArgumentCaptor.forClass(ClientChannelInfo.class); + doNothing().when(this.messagingProcessor).registerConsumer(any(), + anyString(), + channelInfoArgumentCaptor.capture(), + any(), + any(), + any(), + subscriptionDatasArgumentCaptor.capture() + ); + + + HeartbeatResponse response = this.sendConsumerHeartbeat(context); + assertEquals(Code.OK, response.getStatus().getCode()); + + ClientChannelInfo clientChannelInfo = channelInfoArgumentCaptor.getValue(); + assertClientChannelInfo(clientChannelInfo, CONSUMER_GROUP); + + SubscriptionData data = subscriptionDatasArgumentCaptor.getValue().stream().findAny().get(); + assertEquals("TAG", data.getExpressionType()); + assertEquals("tag", data.getSubString()); + } + + protected void assertClientChannelInfo(ClientChannelInfo clientChannelInfo, String group) { + assertEquals(LanguageCode.JAVA, clientChannelInfo.getLanguage()); + assertEquals(CLIENT_ID, clientChannelInfo.getClientId()); + assertTrue(clientChannelInfo.getChannel() instanceof GrpcClientChannel); + GrpcClientChannel channel = (GrpcClientChannel) clientChannelInfo.getChannel(); + assertEquals(REMOTE_ADDR, channel.getRemoteAddress()); + assertEquals(LOCAL_ADDR, channel.getLocalAddress()); + assertEquals(group, channel.getGroup()); + } + + @Test + public void testProducerNotifyClientTermination() throws Throwable { + Context context = createContext(); + + when(this.grpcClientSettingsManager.removeClientSettings(eq(CLIENT_ID))).thenReturn(Settings.newBuilder() + .setClientType(ClientType.PRODUCER) + .setPublishing(Publishing.newBuilder() + .addTopics(Resource.newBuilder().setName(TOPIC).build()) + .build()) + .build()); + ArgumentCaptor channelInfoArgumentCaptor = ArgumentCaptor.forClass(ClientChannelInfo.class); + doNothing().when(this.messagingProcessor).unRegisterProducer(any(), anyString(), channelInfoArgumentCaptor.capture()); + + this.sendProducerTelemetry(context); + this.sendProducerHeartbeat(context); + + NotifyClientTerminationResponse response = this.clientActivity.notifyClientTermination( + context, + NotifyClientTerminationRequest.newBuilder() + .build() + ).get(); + + assertEquals(Code.OK, response.getStatus().getCode()); + ClientChannelInfo clientChannelInfo = channelInfoArgumentCaptor.getValue(); + assertClientChannelInfo(clientChannelInfo, TOPIC); + } + + @Test + public void testConsumerNotifyClientTermination() throws Throwable { + Context context = createContext(); + + when(this.grpcClientSettingsManager.removeClientSettings(eq(CLIENT_ID))).thenReturn(Settings.newBuilder() + .setClientType(ClientType.PUSH_CONSUMER) + .build()); + ArgumentCaptor channelInfoArgumentCaptor = ArgumentCaptor.forClass(ClientChannelInfo.class); + doNothing().when(this.messagingProcessor).unRegisterConsumer(any(), anyString(), channelInfoArgumentCaptor.capture()); + + this.sendConsumerTelemetry(context); + this.sendConsumerHeartbeat(context); + + NotifyClientTerminationResponse response = this.clientActivity.notifyClientTermination( + context, + NotifyClientTerminationRequest.newBuilder() + .setGroup(Resource.newBuilder().setName(CONSUMER_GROUP).build()) + .build() + ).get(); + + assertEquals(Code.OK, response.getStatus().getCode()); + ClientChannelInfo clientChannelInfo = channelInfoArgumentCaptor.getValue(); + assertClientChannelInfo(clientChannelInfo, CONSUMER_GROUP); + } + + protected CompletableFuture sendClientTelemetry(Context ctx, Settings settings) { + when(grpcClientSettingsManager.getClientSettings(any())).thenReturn(settings); + + CompletableFuture future = new CompletableFuture<>(); + StreamObserver responseObserver = new StreamObserver() { + @Override + public void onNext(TelemetryCommand value) { + future.complete(value); + } + + @Override + public void onError(Throwable t) { + + } + + @Override public void onCompleted() { + + } + }; + StreamObserver requestObserver = this.clientActivity.telemetry( + ctx, + responseObserver + ); + requestObserver.onNext(TelemetryCommand.newBuilder() + .setSettings(settings) + .build()); + return future; + } +} \ No newline at end of file diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/ForwardMessageToDLQActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/ForwardMessageToDLQActivityTest.java new file mode 100644 index 0000000000..c6153d7133 --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/ForwardMessageToDLQActivityTest.java @@ -0,0 +1,64 @@ +/* + * 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.proxy.grpc.v2.producer; + +import apache.rocketmq.v2.Code; +import apache.rocketmq.v2.ForwardMessageToDeadLetterQueueRequest; +import apache.rocketmq.v2.ForwardMessageToDeadLetterQueueResponse; +import apache.rocketmq.v2.Resource; +import java.util.concurrent.CompletableFuture; +import org.apache.rocketmq.common.message.MessageClientIDSetter; +import org.apache.rocketmq.common.protocol.ResponseCode; +import org.apache.rocketmq.proxy.grpc.v2.BaseActivityTest; +import org.apache.rocketmq.remoting.protocol.RemotingCommand; +import org.junit.Before; +import org.junit.Test; + +import static org.junit.Assert.*; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.when; + +public class ForwardMessageToDLQActivityTest extends BaseActivityTest { + + private ForwardMessageToDLQActivity forwardMessageToDLQActivity; + + @Before + public void before() throws Throwable { + super.before(); + this.forwardMessageToDLQActivity = new ForwardMessageToDLQActivity(this.messagingProcessor, this.grpcClientSettingsManager); + } + + @Test + public void testForwardMessageToDeadLetterQueue() throws Throwable { + when(this.messagingProcessor.forwardMessageToDeadLetterQueue(any(), any(), anyString(), anyString(), anyString())) + .thenReturn(CompletableFuture.completedFuture(RemotingCommand.createResponseCommand(ResponseCode.SUCCESS, ""))); + + ForwardMessageToDeadLetterQueueResponse response = this.forwardMessageToDLQActivity.forwardMessageToDeadLetterQueue( + createContext(), + ForwardMessageToDeadLetterQueueRequest.newBuilder() + .setTopic(Resource.newBuilder().setName("topic").build()) + .setGroup(Resource.newBuilder().setName("group").build()) + .setMessageId(MessageClientIDSetter.createUniqID()) + .setReceiptHandle(buildReceiptHandle("topic", System.currentTimeMillis(), 3000)) + .build() + ).get(); + + assertEquals(Code.OK, response.getStatus().getCode()); + } +} \ No newline at end of file diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/transaction/EndTransactionActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/transaction/EndTransactionActivityTest.java new file mode 100644 index 0000000000..6709ae05ff --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/transaction/EndTransactionActivityTest.java @@ -0,0 +1,101 @@ +/* + * 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.proxy.grpc.v2.transaction; + +import apache.rocketmq.v2.Code; +import apache.rocketmq.v2.EndTransactionRequest; +import apache.rocketmq.v2.EndTransactionResponse; +import apache.rocketmq.v2.Resource; +import apache.rocketmq.v2.TransactionResolution; +import apache.rocketmq.v2.TransactionSource; +import java.util.Arrays; +import java.util.Collection; +import org.apache.rocketmq.common.message.MessageClientIDSetter; +import org.apache.rocketmq.proxy.grpc.v2.BaseActivityTest; +import org.apache.rocketmq.proxy.processor.TransactionStatus; +import org.apache.rocketmq.proxy.service.transaction.TransactionId; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.mockito.ArgumentCaptor; + +import static org.junit.Assert.assertEquals; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doNothing; + +@RunWith(Parameterized.class) +public class EndTransactionActivityTest extends BaseActivityTest { + + private EndTransactionActivity endTransactionActivity; + private TransactionResolution resolution; + private TransactionSource source; + private TransactionStatus transactionStatus; + private Boolean fromTransactionCheck; + + public EndTransactionActivityTest(TransactionResolution resolution, TransactionSource source, + TransactionStatus transactionStatus, Boolean fromTransactionCheck) { + this.resolution = resolution; + this.source = source; + this.transactionStatus = transactionStatus; + this.fromTransactionCheck = fromTransactionCheck; + } + + @Before + public void before() throws Throwable { + super.before(); + this.endTransactionActivity = new EndTransactionActivity(this.messagingProcessor, this.grpcClientSettingsManager); + } + + @Test + public void testEndTransaction() throws Throwable { + ArgumentCaptor transactionStatusCaptor = ArgumentCaptor.forClass(TransactionStatus.class); + ArgumentCaptor fromTransactionCheckCaptor = ArgumentCaptor.forClass(Boolean.class); + doNothing().when(this.messagingProcessor).endTransaction(any(), any(), anyString(), anyString(), + transactionStatusCaptor.capture(), + fromTransactionCheckCaptor.capture()); + + EndTransactionResponse response = this.endTransactionActivity.endTransaction( + createContext(), + EndTransactionRequest.newBuilder() + .setResolution(resolution) + .setTopic(Resource.newBuilder().setName("topic").build()) + .setMessageId(MessageClientIDSetter.createUniqID()) + .setTransactionId( + TransactionId.genByBrokerTransactionId("brokerName", "0", 0, 0) + .getProxyTransactionId()) + .setSource(source) + .build() + ).get(); + + assertEquals(Code.OK, response.getStatus().getCode()); + assertEquals(transactionStatus, transactionStatusCaptor.getValue()); + assertEquals(fromTransactionCheck, fromTransactionCheckCaptor.getValue()); + } + + @Parameterized.Parameters + public static Collection parameters() { + Object[][] p = new Object[][]{ + {TransactionResolution.COMMIT, TransactionSource.SOURCE_CLIENT, TransactionStatus.COMMIT, false}, + {TransactionResolution.ROLLBACK, TransactionSource.SOURCE_SERVER_CHECK, TransactionStatus.ROLLBACK, true}, + {TransactionResolution.TRANSACTION_RESOLUTION_UNSPECIFIED, TransactionSource.SOURCE_SERVER_CHECK, TransactionStatus.UNKNOWN, true}, + }; + return Arrays.asList(p); + } +} \ No newline at end of file diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/BaseServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/BaseServiceTest.java new file mode 100644 index 0000000000..f6ca31ff84 --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/BaseServiceTest.java @@ -0,0 +1,83 @@ +/* + * 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.proxy.service; + +import java.util.HashMap; +import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.common.MixAll; +import org.apache.rocketmq.common.protocol.ResponseCode; +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.InitConfigAndLoggerTest; +import org.apache.rocketmq.proxy.service.mqclient.MQClientAPIExt; +import org.apache.rocketmq.proxy.service.mqclient.MQClientAPIFactory; +import org.apache.rocketmq.proxy.service.route.MessageQueueView; +import org.apache.rocketmq.proxy.service.route.TopicRouteService; +import org.assertj.core.util.Lists; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.runner.RunWith; +import org.mockito.junit.MockitoJUnitRunner; + +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +@Ignore +@RunWith(MockitoJUnitRunner.Silent.class) +public class BaseServiceTest extends InitConfigAndLoggerTest { + + protected TopicRouteService topicRouteService; + protected MQClientAPIFactory mqClientAPIFactory; + protected MQClientAPIExt mqClientAPIExt; + + protected static final String ERR_TOPIC = "errTopic"; + protected static final String TOPIC = "topic"; + protected static final String GROUP = "group"; + protected static final String BROKER_NAME = "broker"; + protected static final String CLUSTER_NAME = "cluster"; + protected static final String BROKER_ADDR = "127.0.0.1:10911"; + + protected final TopicRouteData topicRouteData = new TopicRouteData(); + protected final QueueData queueData = new QueueData(); + protected final BrokerData brokerData = new BrokerData(); + + @Before + public void before() throws Throwable { + super.before(); + + topicRouteService = mock(TopicRouteService.class); + mqClientAPIFactory = mock(MQClientAPIFactory.class); + mqClientAPIExt = mock(MQClientAPIExt.class); + when(mqClientAPIFactory.getClient()).thenReturn(mqClientAPIExt); + + queueData.setBrokerName(BROKER_NAME); + topicRouteData.setQueueDatas(Lists.newArrayList(queueData)); + brokerData.setCluster(CLUSTER_NAME); + brokerData.setBrokerName(BROKER_NAME); + HashMap brokerAddrs = new HashMap<>(); + brokerAddrs.put(MixAll.MASTER_ID, BROKER_ADDR); + brokerData.setBrokerAddrs(brokerAddrs); + topicRouteData.setBrokerDatas(Lists.newArrayList(brokerData)); + + when(this.topicRouteService.getAllMessageQueueView(eq(ERR_TOPIC))).thenThrow(new MQClientException(ResponseCode.TOPIC_NOT_EXIST, "")); + when(this.topicRouteService.getAllMessageQueueView(eq(TOPIC))).thenReturn(new MessageQueueView(TOPIC, topicRouteData)); + when(this.topicRouteService.getAllMessageQueueView(eq(CLUSTER_NAME))).thenReturn(new MessageQueueView(CLUSTER_NAME, topicRouteData)); + } +} diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/metadata/ClusterMetadataServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/metadata/ClusterMetadataServiceTest.java new file mode 100644 index 0000000000..7eeb72060e --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/metadata/ClusterMetadataServiceTest.java @@ -0,0 +1,73 @@ +/* + * 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.proxy.service.metadata; + +import java.util.HashMap; +import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.common.attribute.TopicMessageType; +import org.apache.rocketmq.common.protocol.ResponseCode; +import org.apache.rocketmq.common.statictopic.TopicConfigAndQueueMapping; +import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig; +import org.apache.rocketmq.proxy.config.ConfigurationManager; +import org.apache.rocketmq.proxy.service.BaseServiceTest; +import org.apache.rocketmq.proxy.service.route.MessageQueueView; +import org.junit.Before; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.when; + +public class ClusterMetadataServiceTest extends BaseServiceTest { + + private ClusterMetadataService clusterMetadataService; + + @Before + public void before() throws Throwable { + super.before(); + ConfigurationManager.getProxyConfig().setRocketMQClusterName(CLUSTER_NAME); + + TopicConfigAndQueueMapping topicConfigAndQueueMapping = new TopicConfigAndQueueMapping(); + topicConfigAndQueueMapping.setAttributes(new HashMap<>()); + topicConfigAndQueueMapping.setTopicMessageType(TopicMessageType.NORMAL); + when(this.mqClientAPIExt.getTopicConfig(anyString(), eq(TOPIC), anyLong())).thenReturn(topicConfigAndQueueMapping); + + when(this.mqClientAPIExt.getSubscriptionGroupConfig(anyString(), eq(GROUP), anyLong())).thenReturn(new SubscriptionGroupConfig()); + + this.clusterMetadataService = new ClusterMetadataService(this.topicRouteService, this.mqClientAPIFactory); + } + + @Test + public void testGetTopicMessageType() { + assertEquals(TopicMessageType.UNSPECIFIED, this.clusterMetadataService.getTopicMessageType(ERR_TOPIC)); + assertEquals(1, this.clusterMetadataService.topicConfigCache.asMap().size()); + assertEquals(TopicMessageType.UNSPECIFIED, this.clusterMetadataService.getTopicMessageType(ERR_TOPIC)); + + assertEquals(TopicMessageType.NORMAL, this.clusterMetadataService.getTopicMessageType(TOPIC)); + assertEquals(2, this.clusterMetadataService.topicConfigCache.asMap().size()); + } + + @Test + public void testGetSubscriptionGroupConfig() { + assertNotNull(this.clusterMetadataService.getSubscriptionGroupConfig(GROUP)); + assertEquals(1, this.clusterMetadataService.subscriptionGroupConfigCache.asMap().size()); + } +} \ No newline at end of file diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/route/ClusterTopicRouteServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/route/ClusterTopicRouteServiceTest.java new file mode 100644 index 0000000000..2d5f64de37 --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/route/ClusterTopicRouteServiceTest.java @@ -0,0 +1,69 @@ +/* + * 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.proxy.service.route; + +import com.google.common.net.HostAndPort; +import java.util.ArrayList; +import java.util.List; +import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.common.MixAll; +import org.apache.rocketmq.common.protocol.ResponseCode; +import org.apache.rocketmq.proxy.common.Address; +import org.apache.rocketmq.proxy.service.BaseServiceTest; +import org.assertj.core.util.Lists; +import org.junit.Before; +import org.junit.Test; + +import static org.assertj.core.api.Assertions.catchThrowableOfType; +import static org.junit.Assert.*; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.when; + +public class ClusterTopicRouteServiceTest extends BaseServiceTest { + + private ClusterTopicRouteService topicRouteService; + + @Before + public void before() throws Throwable { + super.before(); + this.topicRouteService = new ClusterTopicRouteService(this.mqClientAPIFactory); + + when(this.mqClientAPIExt.getTopicRouteInfoFromNameServer(eq(TOPIC), anyLong())).thenReturn(topicRouteData); + when(this.mqClientAPIExt.getTopicRouteInfoFromNameServer(eq(ERR_TOPIC), anyLong())).thenThrow(new MQClientException(ResponseCode.TOPIC_NOT_EXIST, "")); + } + + @Test + public void testGetCurrentMessageQueueView() throws Throwable { + MQClientException exception = catchThrowableOfType(() -> this.topicRouteService.getCurrentMessageQueueView(ERR_TOPIC), MQClientException.class); + assertTrue(TopicRouteHelper.isTopicNotExistError(exception)); + assertEquals(1, this.topicRouteService.topicCache.asMap().size()); + + assertNotNull(this.topicRouteService.getCurrentMessageQueueView(TOPIC)); + assertEquals(2, this.topicRouteService.topicCache.asMap().size()); + } + + @Test + public void testGetTopicRouteForProxy() throws Throwable { + List
addressList = Lists.newArrayList(new Address(Address.AddressScheme.IPv4, HostAndPort.fromParts("127.0.0.1", 8888))); + ProxyTopicRouteData proxyTopicRouteData = this.topicRouteService.getTopicRouteForProxy(addressList, TOPIC); + + assertEquals(1, proxyTopicRouteData.getBrokerDatas().size()); + assertEquals(addressList, proxyTopicRouteData.getBrokerDatas().get(0).getBrokerAddrs().get(MixAll.MASTER_ID)); + } +} \ No newline at end of file diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/route/LocalTopicRouteServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/route/LocalTopicRouteServiceTest.java new file mode 100644 index 0000000000..709d6cc04c --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/route/LocalTopicRouteServiceTest.java @@ -0,0 +1,102 @@ +/* + * 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.proxy.service.route; + +import com.google.common.net.HostAndPort; +import java.util.ArrayList; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import org.apache.rocketmq.broker.BrokerController; +import org.apache.rocketmq.broker.topic.TopicConfigManager; +import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.common.BrokerConfig; +import org.apache.rocketmq.common.MixAll; +import org.apache.rocketmq.common.TopicConfig; +import org.apache.rocketmq.common.constant.PermName; +import org.apache.rocketmq.common.protocol.ResponseCode; +import org.apache.rocketmq.proxy.common.Address; +import org.apache.rocketmq.proxy.config.ConfigurationManager; +import org.apache.rocketmq.proxy.service.BaseServiceTest; +import org.assertj.core.util.Lists; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mock; + +import static org.junit.Assert.assertEquals; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.when; + +public class LocalTopicRouteServiceTest extends BaseServiceTest { + + private static final String LOCAL_BROKER_NAME = "localBroker"; + private static final String LOCAL_CLUSTER_NAME = "localCluster"; + private static final String LOCAL_HOST = "127.0.0.2"; + private static final int LOCAL_PORT = 10911; + private static final String LOCAL_ADDR = LOCAL_HOST + ":" + LOCAL_PORT; + @Mock + private BrokerController brokerController; + @Mock + private TopicConfigManager topicConfigManager; + private ConcurrentMap topicConfigTable = new ConcurrentHashMap<>(); + private BrokerConfig brokerConfig = new BrokerConfig(); + private LocalTopicRouteService topicRouteService; + + @Before + public void before() throws Throwable { + super.before(); + this.brokerConfig.setBrokerName(LOCAL_BROKER_NAME); + this.brokerConfig.setBrokerClusterName(LOCAL_CLUSTER_NAME); + + when(this.brokerController.getBrokerAddr()).thenReturn(LOCAL_ADDR); + when(this.brokerController.getBrokerConfig()).thenReturn(this.brokerConfig); + when(this.brokerController.getTopicConfigManager()).thenReturn(this.topicConfigManager); + when(this.topicConfigManager.getTopicConfigTable()).thenReturn(this.topicConfigTable); + + this.topicRouteService = new LocalTopicRouteService(this.brokerController, this.mqClientAPIFactory); + + when(this.mqClientAPIExt.getTopicRouteInfoFromNameServer(eq(TOPIC), anyLong())).thenReturn(topicRouteData); + when(this.mqClientAPIExt.getTopicRouteInfoFromNameServer(eq(ERR_TOPIC), anyLong())).thenThrow(new MQClientException(ResponseCode.TOPIC_NOT_EXIST, "")); + } + + @Test + public void testGetCurrentMessageQueueView() throws Throwable { + this.topicConfigTable.put(TOPIC, new TopicConfig(TOPIC, 3, 2, PermName.PERM_WRITE | PermName.PERM_READ)); + MessageQueueView messageQueueView = this.topicRouteService.getCurrentMessageQueueView(TOPIC); + assertEquals(3, messageQueueView.getReadSelector().getQueues().size()); + assertEquals(2, messageQueueView.getWriteSelector().getQueues().size()); + assertEquals(1, messageQueueView.getReadSelector().getBrokerActingQueues().size()); + assertEquals(1, messageQueueView.getWriteSelector().getBrokerActingQueues().size()); + + assertEquals(LOCAL_ADDR, messageQueueView.getReadSelector().selectOne(true).getBrokerAddr()); + assertEquals(LOCAL_BROKER_NAME, messageQueueView.getReadSelector().selectOne(true).getBrokerName()); + assertEquals(messageQueueView.getReadSelector().selectOne(true), messageQueueView.getWriteSelector().selectOne(true)); + } + + @Test + public void testGetTopicRouteForProxy() throws Throwable { + ProxyTopicRouteData proxyTopicRouteData = this.topicRouteService.getTopicRouteForProxy(new ArrayList<>(), TOPIC); + + assertEquals(1, proxyTopicRouteData.getBrokerDatas().size()); + assertEquals( + Lists.newArrayList(new Address(Address.AddressScheme.IPv4, HostAndPort.fromParts( + HostAndPort.fromString(BROKER_ADDR).getHost(), + ConfigurationManager.getProxyConfig().getGrpcServerPort()))), + proxyTopicRouteData.getBrokerDatas().get(0).getBrokerAddrs().get(MixAll.MASTER_ID)); + } +} \ No newline at end of file diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelectorTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelectorTest.java new file mode 100644 index 0000000000..a59c4cf20a --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelectorTest.java @@ -0,0 +1,84 @@ +/* + * 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.proxy.service.route; + +import org.apache.rocketmq.common.constant.PermName; +import org.apache.rocketmq.proxy.service.BaseServiceTest; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class MessageQueueSelectorTest extends BaseServiceTest { + + @Test + public void testReadMessageQueue() { + queueData.setPerm(PermName.PERM_READ); + queueData.setReadQueueNums(0); + MessageQueueSelector messageQueueSelector = new MessageQueueSelector(new TopicRouteWrapper(topicRouteData, TOPIC), true); + assertTrue(messageQueueSelector.getQueues().isEmpty()); + + queueData.setPerm(PermName.PERM_READ); + queueData.setReadQueueNums(3); + messageQueueSelector = new MessageQueueSelector(new TopicRouteWrapper(topicRouteData, TOPIC), true); + assertEquals(3, messageQueueSelector.getQueues().size()); + assertEquals(1, messageQueueSelector.getBrokerActingQueues().size()); + for (int i = 0; i < messageQueueSelector.getQueues().size(); i++) { + SelectableMessageQueue messageQueue = messageQueueSelector.getQueues().get(i); + assertEquals(i, messageQueue.getQueueId()); + } + + SelectableMessageQueue brokerQueue = messageQueueSelector.getQueueByBrokerName(BROKER_NAME); + assertEquals(brokerQueue, messageQueueSelector.getBrokerActingQueues().get(0)); + assertEquals(brokerQueue, messageQueueSelector.selectOne(true)); + assertEquals(brokerQueue, messageQueueSelector.selectOneByIndex(3, true)); + + SelectableMessageQueue queue = messageQueueSelector.selectOne(false); + messageQueueSelector.selectOne(false); + messageQueueSelector.selectOne(false); + assertEquals(queue, messageQueueSelector.selectOne(false)); + } + + @Test + public void testWriteMessageQueue() { + queueData.setPerm(PermName.PERM_WRITE); + queueData.setReadQueueNums(0); + MessageQueueSelector messageQueueSelector = new MessageQueueSelector(new TopicRouteWrapper(topicRouteData, TOPIC), false); + assertTrue(messageQueueSelector.getQueues().isEmpty()); + + queueData.setPerm(PermName.PERM_WRITE); + queueData.setWriteQueueNums(3); + messageQueueSelector = new MessageQueueSelector(new TopicRouteWrapper(topicRouteData, TOPIC), false); + assertEquals(3, messageQueueSelector.getQueues().size()); + assertEquals(1, messageQueueSelector.getBrokerActingQueues().size()); + for (int i = 0; i < messageQueueSelector.getQueues().size(); i++) { + SelectableMessageQueue messageQueue = messageQueueSelector.getQueues().get(i); + assertEquals(i, messageQueue.getQueueId()); + } + + SelectableMessageQueue brokerQueue = messageQueueSelector.getQueueByBrokerName(BROKER_NAME); + assertEquals(brokerQueue, messageQueueSelector.getBrokerActingQueues().get(0)); + assertEquals(brokerQueue, messageQueueSelector.selectOne(true)); + assertEquals(brokerQueue, messageQueueSelector.selectOneByIndex(3, true)); + + SelectableMessageQueue queue = messageQueueSelector.selectOne(false); + messageQueueSelector.selectOne(false); + messageQueueSelector.selectOne(false); + assertEquals(queue, messageQueueSelector.selectOne(false)); + } +} \ No newline at end of file diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionServiceTest.java new file mode 100644 index 0000000000..517838f42c --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionServiceTest.java @@ -0,0 +1,141 @@ +/* + * 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.proxy.service.transaction; + +import java.time.Duration; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import org.apache.rocketmq.broker.client.ProducerManager; +import org.apache.rocketmq.common.protocol.heartbeat.HeartbeatData; +import org.apache.rocketmq.common.protocol.heartbeat.ProducerData; +import org.apache.rocketmq.proxy.config.ConfigurationManager; +import org.apache.rocketmq.proxy.service.BaseServiceTest; +import org.apache.rocketmq.proxy.service.route.MessageQueueView; +import org.assertj.core.util.Lists; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +import static org.awaitility.Awaitility.await; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.when; + + +public class ClusterTransactionServiceTest extends BaseServiceTest { + + @Mock + private ProducerManager producerManager; + + private ClusterTransactionService clusterTransactionService; + + @Before + public void before() throws Throwable { + super.before(); + this.clusterTransactionService = new ClusterTransactionService(this.topicRouteService, this.producerManager, null, + this.mqClientAPIFactory); + + MessageQueueView messageQueueView = new MessageQueueView(TOPIC, topicRouteData); + when(this.topicRouteService.getAllMessageQueueView(anyString())) + .thenReturn(messageQueueView); + + when(mqClientAPIFactory.getClient()).thenReturn(mqClientAPIExt); + } + + @Test + public void testAddTransactionSubscription() { + this.clusterTransactionService.addTransactionSubscription(GROUP, TOPIC); + + assertEquals(1, this.clusterTransactionService.getGroupClusterData().size()); + assertEquals(CLUSTER_NAME, this.clusterTransactionService.getGroupClusterData().get(GROUP).stream().findAny().get().getCluster()); + } + + @Test + public void testAddTransactionSubscriptionTopicList() { + this.clusterTransactionService.addTransactionSubscription(GROUP, Lists.newArrayList(TOPIC + 1, TOPIC + 2)); + + assertEquals(1, this.clusterTransactionService.getGroupClusterData().size()); + assertEquals(CLUSTER_NAME, this.clusterTransactionService.getGroupClusterData().get(GROUP).stream().findAny().get().getCluster()); + } + + @Test + public void testReplaceTransactionSubscription() { + this.clusterTransactionService.addTransactionSubscription(GROUP, TOPIC); + + assertEquals(1, this.clusterTransactionService.getGroupClusterData().size()); + assertEquals(CLUSTER_NAME, this.clusterTransactionService.getGroupClusterData().get(GROUP).stream().findAny().get().getCluster()); + + this.brokerData.setCluster(CLUSTER_NAME + 1); + this.clusterTransactionService.replaceTransactionSubscription(GROUP, Lists.newArrayList(TOPIC + 1)); + assertEquals(1, this.clusterTransactionService.getGroupClusterData().size()); + assertEquals(CLUSTER_NAME + 1, this.clusterTransactionService.getGroupClusterData().get(GROUP).stream().findAny().get().getCluster()); + } + + @Test + public void testUnSubscribeAllTransactionTopic() { + this.clusterTransactionService.addTransactionSubscription(GROUP, TOPIC); + this.clusterTransactionService.unSubscribeAllTransactionTopic(GROUP); + + assertEquals(0, this.clusterTransactionService.getGroupClusterData().size()); + } + + @Test + public void testScanProducerHeartBeat() throws Exception { + ConfigurationManager.getProxyConfig().setTransactionHeartbeatBatchNum(2); + this.clusterTransactionService.start(); + Set groupSet = new HashSet<>(); + + for (int i = 0; i < 3; i++) { + groupSet.add(GROUP + i); + this.clusterTransactionService.addTransactionSubscription(GROUP + i, TOPIC); + } + + ArgumentCaptor brokerAddrArgumentCaptor = ArgumentCaptor.forClass(String.class); + ArgumentCaptor heartbeatDataArgumentCaptor = ArgumentCaptor.forClass(HeartbeatData.class); + doNothing().when(mqClientAPIExt).sendHeartbeatOneway( + brokerAddrArgumentCaptor.capture(), + heartbeatDataArgumentCaptor.capture(), + anyLong() + ); + + this.clusterTransactionService.scanProducerHeartBeat(); + + await().atMost(Duration.ofSeconds(1)).until(() -> brokerAddrArgumentCaptor.getAllValues().size() == 2); + + assertEquals(Lists.newArrayList(BROKER_ADDR, BROKER_ADDR), brokerAddrArgumentCaptor.getAllValues()); + List heartbeatDataList = heartbeatDataArgumentCaptor.getAllValues(); + assertEquals(2, heartbeatDataList.get(0).getProducerDataSet().size()); + for (ProducerData producerData : heartbeatDataList.get(0).getProducerDataSet()) { + groupSet.remove(producerData.getGroupName()); + } + + assertEquals(1, heartbeatDataList.get(1).getProducerDataSet().size()); + for (ProducerData producerData : heartbeatDataList.get(1).getProducerDataSet()) { + groupSet.remove(producerData.getGroupName()); + } + + assertTrue(groupSet.isEmpty()); + } +} \ No newline at end of file