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] v2 test cases

Browse Source
This commit is contained in:
kaiyi.lk
2022-07-13 11:29:17 +08:00
committed by zhouxiang
parent 46ace35380
commit 74cac24c67
9 changed files with 244 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/BaseServiceTest.java
+23
View File
@@ -16,6 +16,8 @@
*/
package org.apache.rocketmq.proxy.grpc.v2.service.cluster;
import io.grpc.Context;
import io.grpc.Metadata;
import java.net.SocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ThreadLocalRandom;
@@ -32,6 +34,8 @@ import org.apache.rocketmq.proxy.connector.ForwardReadConsumer;
import org.apache.rocketmq.proxy.connector.route.TopicRouteCache;
import org.apache.rocketmq.proxy.connector.ForwardWriteConsumer;
import org.apache.rocketmq.proxy.connector.transaction.TransactionHeartbeatRegisterService;
import org.apache.rocketmq.proxy.grpc.interceptor.InterceptorConstants;
import org.apache.rocketmq.proxy.grpc.v2.service.GrpcClientManager;
import org.apache.rocketmq.remoting.common.RemotingUtil;
import org.junit.Before;
import org.junit.Ignore;
@@ -59,6 +63,13 @@ public abstract class BaseServiceTest extends InitConfigAndLoggerTest {
protected TopicRouteCache topicRouteCache;
@Mock
protected TransactionHeartbeatRegisterService transactionHeartbeatRegisterService;
@Mock
protected GrpcClientManager grpcClientManager;
public static final String REMOTE_ADDRESS = "1.1.1.1";
public static final String LOCAL_ADDRESS = "0.0.0.0";
public static final String LANGUAGE = "JAVA";
public static final String CLIENT_ID = "client-id";
@Before
public void before() throws Throwable {
@@ -70,6 +81,7 @@ public abstract class BaseServiceTest extends InitConfigAndLoggerTest {
when(connectorManager.getTopicRouteCache()).thenReturn(topicRouteCache);
when(connectorManager.getTransactionHeartbeatRegisterService()).thenReturn(transactionHeartbeatRegisterService);
initContext();
beforeEach();
}
@@ -106,4 +118,15 @@ public abstract class BaseServiceTest extends InitConfigAndLoggerTest {
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_POP_CK, handler);
return msg;
}
protected static Context initContext() {
Context ctx = Context.current();
Metadata metadata = new Metadata();
metadata.put(InterceptorConstants.REMOTE_ADDRESS, REMOTE_ADDRESS);
metadata.put(InterceptorConstants.LOCAL_ADDRESS, LOCAL_ADDRESS);
metadata.put(InterceptorConstants.LANGUAGE, LANGUAGE);
metadata.put(InterceptorConstants.CLIENT_ID, CLIENT_ID);
ctx.withValue(InterceptorConstants.METADATA, metadata).attach();
return ctx;
}
}
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ConsumerServiceTest.java
+39 -20
View File
@@ -1,16 +1,19 @@
package org.apache.rocketmq.proxy.grpc.v2.service.cluster;
import apache.rocketmq.v1.AckMessageRequest;
import apache.rocketmq.v1.AckMessageResponse;
import apache.rocketmq.v1.FilterExpression;
import apache.rocketmq.v1.FilterType;
import apache.rocketmq.v1.NackMessageRequest;
import apache.rocketmq.v1.NackMessageResponse;
import apache.rocketmq.v1.Partition;
import apache.rocketmq.v1.ReceiveMessageRequest;
import apache.rocketmq.v1.ReceiveMessageResponse;
import apache.rocketmq.v1.Resource;
import com.google.rpc.Code;
import apache.rocketmq.v2.AckMessageRequest;
import apache.rocketmq.v2.AckMessageResponse;
import apache.rocketmq.v2.ClientSettings;
import apache.rocketmq.v2.Code;
import apache.rocketmq.v2.DeadLetterPolicy;
import apache.rocketmq.v2.FilterExpression;
import apache.rocketmq.v2.FilterType;
import apache.rocketmq.v2.NackMessageRequest;
import apache.rocketmq.v2.NackMessageResponse;
import apache.rocketmq.v2.ReceiveMessageRequest;
import apache.rocketmq.v2.ReceiveMessageResponse;
import apache.rocketmq.v2.Resource;
import apache.rocketmq.v2.Settings;
import apache.rocketmq.v2.Subscription;
import io.grpc.Context;
import java.util.List;
import java.util.concurrent.CompletableFuture;
@@ -49,7 +52,7 @@ public class ConsumerServiceTest extends BaseServiceTest {
@Override
public void beforeEach() throws Throwable {
consumerService = new ConsumerService(this.connectorManager);
consumerService = new ConsumerService(this.connectorManager, this.grpcClientManager);
consumerService.setReadQueueSelector(readQueueSelector);
}
@@ -75,7 +78,7 @@ public class ConsumerServiceTest extends BaseServiceTest {
consumerService.setAckNoMatchedMessageHook((ctx1, request, response, t) -> ackHandler.set(request.getExtraInfo()));
ReceiveMessageResponse response = consumerService.receiveMessage(ctx,
ReceiveMessageRequest.newBuilder()
.setPartition(Partition.newBuilder()
.setMessageQueue(apache.rocketmq.v2.MessageQueue.newBuilder()
.setTopic(Resource.newBuilder()
.setResourceNamespace("namespace")
.setName("topic")
@@ -88,9 +91,9 @@ public class ConsumerServiceTest extends BaseServiceTest {
.build()
).get();
assertEquals(Code.OK.getNumber(), response.getCommon().getStatus().getCode());
assertEquals(Code.OK, response.getStatus().getCode());
assertEquals(1, response.getMessagesCount());
assertEquals("msg1", response.getMessages(0).getSystemAttribute().getMessageId());
assertEquals("msg1", response.getMessages(0).getSystemProperties().getMessageId());
assertEquals(ReceiptHandle.create(messageExtList.get(1)).getReceiptHandle(), ackHandler.get());
}
@@ -112,7 +115,7 @@ public class ConsumerServiceTest extends BaseServiceTest {
.build())
.get();
assertEquals(Code.OK.getNumber(), response.getCommon().getStatus().getCode());
assertEquals(Code.OK, response.getStatus().getCode());
}
@Test
@@ -125,6 +128,9 @@ public class ConsumerServiceTest extends BaseServiceTest {
}).when(producerClient).sendMessageBack(anyString(), any());
when(topicRouteCache.getBrokerAddr(anyString())).thenReturn("brokerAddr");
ClientSettings clientSettings = createClientSettings(3);
when(grpcClientManager.getClientSettings(anyString())).thenReturn(clientSettings);
NackMessageResponse response = consumerService.nackMessage(Context.current(), NackMessageRequest.newBuilder()
.setTopic(Resource.newBuilder()
.setName("topic")
@@ -134,11 +140,10 @@ public class ConsumerServiceTest extends BaseServiceTest {
.build())
.setReceiptHandle(receiptHandle.encode())
.setDeliveryAttempt(3)
.setMaxDeliveryAttempts(3)
.build())
.get();
assertEquals(Code.OK.getNumber(), response.getCommon().getStatus().getCode());
assertEquals(Code.OK, response.getStatus().getCode());
assertEquals(receiptHandle.getCommitLogOffset(), headerRef.get().getOffset().longValue());
}
@@ -154,6 +159,9 @@ public class ConsumerServiceTest extends BaseServiceTest {
}).when(writeConsumerClient).changeInvisibleTimeAsync(anyString(), anyString(), any());
when(topicRouteCache.getBrokerAddr(anyString())).thenReturn("brokerAddr");
ClientSettings clientSettings = createClientSettings(3);
when(grpcClientManager.getClientSettings(anyString())).thenReturn(clientSettings);
NackMessageResponse response = consumerService.nackMessage(Context.current(), NackMessageRequest.newBuilder()
.setTopic(Resource.newBuilder()
.setName("topic")
@@ -163,12 +171,23 @@ public class ConsumerServiceTest extends BaseServiceTest {
.build())
.setReceiptHandle(receiptHandle.encode())
.setDeliveryAttempt(1)
.setMaxDeliveryAttempts(3)
.build())
.get();
assertEquals(Code.OK.getNumber(), response.getCommon().getStatus().getCode());
assertEquals(Code.OK, response.getStatus().getCode());
assertEquals(receiptHandle.getOffset(), headerRef.get().getOffset().longValue());
assertEquals(receiptHandle.encode(), headerRef.get().getExtraInfo());
}
private ClientSettings createClientSettings(int maxDeliveryAttempts) {
return ClientSettings.newBuilder()
.setSettings(Settings.newBuilder()
.setSubscription(Subscription.newBuilder()
.setDeadLetterPolicy(DeadLetterPolicy.newBuilder()
.setMaxDeliveryAttempts(maxDeliveryAttempts)
.build())
.build())
.build())
.build();
}
}
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultReadQueueSelectorTest.java
+3 -4
View File
@@ -1,8 +1,7 @@
package org.apache.rocketmq.proxy.grpc.v2.service.cluster;
import apache.rocketmq.v1.Broker;
import apache.rocketmq.v1.Partition;
import apache.rocketmq.v1.ReceiveMessageRequest;
import apache.rocketmq.v2.Broker;
import apache.rocketmq.v2.ReceiveMessageRequest;
import io.grpc.Context;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.header.PopMessageRequestHeader;
@@ -41,7 +40,7 @@ public class DefaultReadQueueSelectorTest extends BaseServiceTest {
requestHeader.setTopic("readBrokerTopicByName");
SelectableMessageQueue messageQueue = readQueueSelector.select(Context.current(),
ReceiveMessageRequest.newBuilder()
.setPartition(Partition.newBuilder()
.setMessageQueue(apache.rocketmq.v2.MessageQueue.newBuilder()
.setBroker(Broker.newBuilder()
.setName("brokerName")
.build())
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultWriteQueueSelectorTest.java
+22 -45
View File
@@ -1,18 +1,17 @@
package org.apache.rocketmq.proxy.grpc.v2.service.cluster;
import apache.rocketmq.v1.Broker;
import apache.rocketmq.v1.Message;
import apache.rocketmq.v1.Partition;
import apache.rocketmq.v1.Resource;
import apache.rocketmq.v1.SendMessageRequest;
import apache.rocketmq.v1.SystemAttribute;
import apache.rocketmq.v2.Broker;
import apache.rocketmq.v2.Message;
import apache.rocketmq.v2.Resource;
import apache.rocketmq.v2.SendMessageRequest;
import apache.rocketmq.v2.SystemProperties;
import com.google.protobuf.ByteString;
import io.grpc.Context;
import java.nio.charset.StandardCharsets;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.proxy.connector.route.SelectableMessageQueue;
import org.apache.rocketmq.proxy.grpc.v1.adapter.GrpcConverter;
import org.apache.rocketmq.proxy.grpc.v2.adapter.GrpcConverter;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
@@ -23,6 +22,8 @@ import static org.mockito.Mockito.when;
public class DefaultWriteQueueSelectorTest extends BaseServiceTest {
private static final String TOPIC_NAME = "namespace%topic";
@Override
public void beforeEach() throws Throwable {
SelectableMessageQueue queue = new SelectableMessageQueue(
@@ -44,40 +45,16 @@ public class DefaultWriteQueueSelectorTest extends BaseServiceTest {
.thenReturn(queue);
}
@Test
public void testSendOrderMessageWithShardingKey() {
SendMessageRequest request = SendMessageRequest.newBuilder()
.setMessage(Message.newBuilder()
.setTopic(Resource.newBuilder()
.setResourceNamespace("namespace")
.setName("topic")
.build())
.putUserAttribute(MessageConst.PROPERTY_SHARDING_KEY, "key")
.setSystemAttribute(SystemAttribute.newBuilder()
.setMessageId("msgId")
.build())
.setBody(ByteString.copyFrom("hello", StandardCharsets.UTF_8))
.build())
.build();
WriteQueueSelector queueSelector = new DefaultWriteQueueSelector(this.topicRouteCache);
SelectableMessageQueue queue = queueSelector.selectQueue(Context.current(), request,
GrpcConverter.buildSendMessageRequestHeader(request),
GrpcConverter.buildMessage(request.getMessage()));
assertEquals("selectOrderQueue", queue.getBrokerName());
assertEquals("selectOrderQueueAddr", queue.getBrokerAddr());
}
@Test
public void selectWithShardingKey() {
SendMessageRequest request = SendMessageRequest.newBuilder()
.setMessage(Message.newBuilder()
.addMessages(Message.newBuilder()
.setTopic(Resource.newBuilder()
.setResourceNamespace("namespace")
.setName("topic")
.build())
.putUserAttribute(MessageConst.PROPERTY_SHARDING_KEY, "key")
.setSystemAttribute(SystemAttribute.newBuilder()
.putUserProperties(MessageConst.PROPERTY_SHARDING_KEY, "key")
.setSystemProperties(SystemProperties.newBuilder()
.setMessageId("msgId")
.build())
.setBody(ByteString.copyFrom("hello", StandardCharsets.UTF_8))
@@ -85,8 +62,8 @@ public class DefaultWriteQueueSelectorTest extends BaseServiceTest {
.build();
WriteQueueSelector queueSelector = new DefaultWriteQueueSelector(this.topicRouteCache);
SelectableMessageQueue queue = queueSelector.selectQueue(Context.current(), request,
GrpcConverter.buildSendMessageRequestHeader(request),
GrpcConverter.buildMessage(request.getMessage()));
GrpcConverter.buildSendMessageRequestHeader(request, TOPIC_NAME),
GrpcConverter.buildMessage(request.getMessagesList(), TOPIC_NAME));
assertEquals("selectOrderQueue", queue.getBrokerName());
assertEquals("selectOrderQueueAddr", queue.getBrokerAddr());
@@ -95,12 +72,12 @@ public class DefaultWriteQueueSelectorTest extends BaseServiceTest {
@Test
public void selectNormalQueue() {
SendMessageRequest request = SendMessageRequest.newBuilder()
.setMessage(Message.newBuilder()
.addMessages(Message.newBuilder()
.setTopic(Resource.newBuilder()
.setResourceNamespace("namespace")
.setName("topic")
.build())
.setSystemAttribute(SystemAttribute.newBuilder()
.setSystemProperties(SystemProperties.newBuilder()
.setMessageId("msgId")
.build())
.setBody(ByteString.copyFrom("hello", StandardCharsets.UTF_8))
@@ -108,8 +85,8 @@ public class DefaultWriteQueueSelectorTest extends BaseServiceTest {
.build();
WriteQueueSelector queueSelector = new DefaultWriteQueueSelector(this.topicRouteCache);
SelectableMessageQueue queue = queueSelector.selectQueue(Context.current(), request,
GrpcConverter.buildSendMessageRequestHeader(request),
GrpcConverter.buildMessage(request.getMessage()));
GrpcConverter.buildSendMessageRequestHeader(request, TOPIC_NAME),
GrpcConverter.buildMessage(request.getMessagesList(), TOPIC_NAME));
assertEquals("selectNormalQueue", queue.getBrokerName());
assertEquals("selectNormalQueueAddr", queue.getBrokerAddr());
@@ -118,17 +95,17 @@ public class DefaultWriteQueueSelectorTest extends BaseServiceTest {
@Test
public void selectTargetQueue() {
SendMessageRequest request = SendMessageRequest.newBuilder()
.setMessage(Message.newBuilder()
.addMessages(Message.newBuilder()
.setTopic(Resource.newBuilder()
.setResourceNamespace("namespace")
.setName("topic")
.build())
.setSystemAttribute(SystemAttribute.newBuilder()
.setSystemProperties(SystemProperties.newBuilder()
.setMessageId("msgId")
.build())
.setBody(ByteString.copyFrom("hello", StandardCharsets.UTF_8))
.build())
.setPartition(Partition.newBuilder()
.setMessageQueue(apache.rocketmq.v2.MessageQueue.newBuilder()
.setBroker(Broker.newBuilder()
.setName("brokerName")
.build())
@@ -136,8 +113,8 @@ public class DefaultWriteQueueSelectorTest extends BaseServiceTest {
.build();
WriteQueueSelector queueSelector = new DefaultWriteQueueSelector(this.topicRouteCache);
SelectableMessageQueue queue = queueSelector.selectQueue(Context.current(), request,
GrpcConverter.buildSendMessageRequestHeader(request),
GrpcConverter.buildMessage(request.getMessage()));
GrpcConverter.buildSendMessageRequestHeader(request, TOPIC_NAME),
GrpcConverter.buildMessage(request.getMessagesList(), TOPIC_NAME));
assertEquals("selectTargetQueue", queue.getBrokerName());
assertEquals("selectTargetQueueAddr", queue.getBrokerAddr());
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ForwardClientServiceTest.java
+58 -66
View File
@@ -1,18 +1,17 @@
package org.apache.rocketmq.proxy.grpc.v2.service.cluster;
import apache.rocketmq.v1.ConsumeMessageType;
import apache.rocketmq.v1.ConsumeModel;
import apache.rocketmq.v1.ConsumePolicy;
import apache.rocketmq.v1.ConsumerData;
import apache.rocketmq.v1.FilterExpression;
import apache.rocketmq.v1.FilterType;
import apache.rocketmq.v1.HeartbeatRequest;
import apache.rocketmq.v1.NotifyClientTerminationRequest;
import apache.rocketmq.v1.ProducerData;
import apache.rocketmq.v1.Resource;
import apache.rocketmq.v1.SubscriptionEntry;
import apache.rocketmq.v2.ClientSettings;
import apache.rocketmq.v2.ClientType;
import apache.rocketmq.v2.FilterExpression;
import apache.rocketmq.v2.FilterType;
import apache.rocketmq.v2.HeartbeatRequest;
import apache.rocketmq.v2.NotifyClientTerminationRequest;
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 io.grpc.Context;
import io.grpc.Metadata;
import io.netty.channel.Channel;
import java.util.ArrayList;
import java.util.List;
@@ -24,67 +23,63 @@ import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.proxy.channel.ChannelManager;
import org.apache.rocketmq.proxy.common.PollResponseManager;
import org.apache.rocketmq.proxy.grpc.v1.adapter.channel.GrpcClientChannel;
import org.apache.rocketmq.proxy.grpc.interceptor.InterceptorConstants;
import org.apache.rocketmq.proxy.grpc.v2.adapter.channel.GrpcClientChannel;
import org.apache.rocketmq.remoting.protocol.LanguageCode;
import org.junit.Test;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.when;
public class ForwardClientServiceTest extends BaseServiceTest {
private ChannelManager channelManager = new ChannelManager();
private PollResponseManager pollResponseManager = new PollResponseManager();
private ForwardClientService clientService;
@Override
public void beforeEach() throws Throwable {
clientService = new ForwardClientService(
this.connectorManager,
Executors.newSingleThreadScheduledExecutor(),
this.channelManager,
this.grpcClientManager,
this.pollResponseManager);
}
@Test
public void testProducerHeartbeat() {
ForwardClientService clientService = new ForwardClientService(
this.connectorManager,
Executors.newSingleThreadScheduledExecutor(),
this.channelManager,
this.pollResponseManager);
Metadata metadata = new Metadata();
metadata.put(InterceptorConstants.LANGUAGE, "JAVA");
metadata.put(InterceptorConstants.REMOTE_ADDRESS, "127.0.0.1:8080");
metadata.put(InterceptorConstants.LOCAL_ADDRESS, "127.0.0.1:8081");
Context ctx = Context.current().withValue(InterceptorConstants.METADATA, metadata);
clientService.heartbeat(ctx, HeartbeatRequest.newBuilder()
.setClientId("clientId")
.setProducerData(ProducerData.newBuilder()
.setGroup(Resource.newBuilder()
.setName("producerGroup")
ClientSettings clientSettings = ClientSettings.newBuilder()
.setClientType(ClientType.PRODUCER)
.setSettings(Settings.newBuilder()
.setPublishing(Publishing.newBuilder()
.addTopics(Resource.newBuilder()
.setName("topic1")
.build())
.addTopics(Resource.newBuilder()
.setName("topic2")
.build())
.build())
.build())
.build());
.build();
when(grpcClientManager.getClientSettings(anyString())).thenReturn(clientSettings);
assertEquals(1, clientService.getProducerManager().getGroupChannelTable().size());
Channel channel = clientService.getProducerManager().findChannel("clientId");
clientService.heartbeat(Context.current(), HeartbeatRequest.newBuilder().build());
assertEquals(2, clientService.getProducerManager().getGroupChannelTable().size());
Channel channel = clientService.getProducerManager().findChannel(CLIENT_ID);
assertNotNull(channel);
assertTrue(channel instanceof GrpcClientChannel);
clientService.notifyClientTermination(ctx, NotifyClientTerminationRequest.newBuilder()
.setClientId("clientId")
.setProducerGroup(Resource.newBuilder()
.setName("producerGroup")
.build())
.build());
clientService.notifyClientTermination(Context.current(), NotifyClientTerminationRequest.newBuilder().build());
assertTrue(clientService.getProducerManager().getGroupChannelTable().isEmpty());
}
@Test
public void testConsumerHeartbeat() {
ForwardClientService clientService = new ForwardClientService(
this.connectorManager,
Executors.newSingleThreadScheduledExecutor(),
this.channelManager,
this.pollResponseManager);
List<SubscriptionEntry> subscriptionEntryList = new ArrayList<>();
subscriptionEntryList.add(SubscriptionEntry.newBuilder()
.setTopic(Resource.newBuilder()
@@ -95,29 +90,27 @@ public class ForwardClientServiceTest extends BaseServiceTest {
.setType(FilterType.TAG)
.build())
.build());
Metadata metadata = new Metadata();
metadata.put(InterceptorConstants.LANGUAGE, "JAVA");
metadata.put(InterceptorConstants.REMOTE_ADDRESS, "127.0.0.1:8080");
metadata.put(InterceptorConstants.LOCAL_ADDRESS, "127.0.0.1:8081");
Context ctx = Context.current().withValue(InterceptorConstants.METADATA, metadata);
clientService.heartbeat(ctx, HeartbeatRequest.newBuilder()
.setClientId("clientId")
.setConsumerData(ConsumerData.newBuilder()
.setGroup(Resource.newBuilder()
.setName("consumerGroup")
ClientSettings clientSettings = ClientSettings.newBuilder()
.setClientType(ClientType.PUSH_CONSUMER)
.setSettings(Settings.newBuilder()
.setSubscription(Subscription.newBuilder()
.addAllSubscriptions(subscriptionEntryList)
.build())
.setConsumeType(ConsumeMessageType.PASSIVE)
.setConsumeModel(ConsumeModel.CLUSTERING)
.setConsumePolicy(ConsumePolicy.RESUME)
.addAllSubscriptions(subscriptionEntryList)
.build())
.build();
when(grpcClientManager.getClientSettings(anyString())).thenReturn(clientSettings);
clientService.heartbeat(Context.current(), HeartbeatRequest.newBuilder()
.setGroup(Resource.newBuilder()
.setName("consumerGroup")
.build())
.build());
ClientChannelInfo clientChannelInfo = clientService.getConsumerManager().findChannel("consumerGroup", "clientId");
ClientChannelInfo clientChannelInfo = clientService.getConsumerManager().findChannel("consumerGroup", CLIENT_ID);
assertNotNull(clientChannelInfo);
assertEquals(LanguageCode.JAVA, clientChannelInfo.getLanguage());
assertEquals("clientId", clientChannelInfo.getClientId());
assertEquals(CLIENT_ID, clientChannelInfo.getClientId());
assertTrue(clientChannelInfo.getChannel() instanceof GrpcClientChannel);
ConsumerGroupInfo consumerGroupInfo = clientService.getConsumerManager().getConsumerGroupInfo("consumerGroup");
assertEquals(MessageModel.CLUSTERING, consumerGroupInfo.getMessageModel());
@@ -127,9 +120,8 @@ public class ForwardClientServiceTest extends BaseServiceTest {
assertEquals("*", consumerGroupInfo.getSubscriptionTable().get("topic").getSubString());
clientService.notifyClientTermination(ctx, NotifyClientTerminationRequest.newBuilder()
.setClientId("clientId")
.setConsumerGroup(Resource.newBuilder()
clientService.notifyClientTermination(Context.current(), NotifyClientTerminationRequest.newBuilder()
.setGroup(Resource.newBuilder()
.setName("consumerGroup")
.build())
.build());
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ProducerServiceTest.java
+27 -14
View File
@@ -16,13 +16,14 @@
*/
package org.apache.rocketmq.proxy.grpc.v2.service.cluster;
import apache.rocketmq.v1.Message;
import apache.rocketmq.v1.Resource;
import apache.rocketmq.v1.SendMessageRequest;
import apache.rocketmq.v1.SendMessageResponse;
import apache.rocketmq.v1.SystemAttribute;
import apache.rocketmq.v2.Code;
import apache.rocketmq.v2.ForwardMessageToDeadLetterQueueRequest;
import apache.rocketmq.v2.Message;
import apache.rocketmq.v2.Resource;
import apache.rocketmq.v2.SendMessageRequest;
import apache.rocketmq.v2.SendMessageResponse;
import apache.rocketmq.v2.SystemProperties;
import com.google.protobuf.ByteString;
import com.google.rpc.Code;
import io.grpc.Context;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
@@ -31,7 +32,7 @@ import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.proxy.connector.route.SelectableMessageQueue;
import org.apache.rocketmq.proxy.grpc.v1.adapter.ProxyException;
import org.apache.rocketmq.proxy.grpc.v2.adapter.ProxyException;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
@@ -46,12 +47,12 @@ import static org.mockito.Mockito.when;
public class ProducerServiceTest extends BaseServiceTest {
private static final SendMessageRequest REQUEST = SendMessageRequest.newBuilder()
.setMessage(Message.newBuilder()
.addMessages(Message.newBuilder()
.setTopic(Resource.newBuilder()
.setResourceNamespace("namespace")
.setName("topic")
.build())
.setSystemAttribute(SystemAttribute.newBuilder()
.setSystemProperties(SystemProperties.newBuilder()
.setMessageId("msgId")
.build())
.setBody(ByteString.copyFrom("hello", StandardCharsets.UTF_8))
@@ -78,8 +79,8 @@ public class ProducerServiceTest extends BaseServiceTest {
try {
SendMessageResponse response = future.get();
assertEquals(Code.OK.getNumber(), response.getCommon().getStatus().getCode());
assertEquals("msgId", response.getMessageId());
assertEquals(Code.OK, response.getStatus().getCode());
assertEquals("msgId", response.getReceipts(0).getMessageId());
} catch (Exception e) {
assertNull(e);
}
@@ -92,12 +93,12 @@ public class ProducerServiceTest extends BaseServiceTest {
producerService.setWriteQueueSelector((ctx, request, requestHeader, message) -> null);
CompletableFuture<SendMessageResponse> future = producerService.sendMessage(Context.current(), SendMessageRequest.newBuilder()
.setMessage(Message.newBuilder()
.addMessages(Message.newBuilder()
.setTopic(Resource.newBuilder()
.setResourceNamespace("namespace")
.setName("topic")
.build())
.setSystemAttribute(SystemAttribute.newBuilder()
.setSystemProperties(SystemProperties.newBuilder()
.setMessageId("msgId")
.build())
.setBody(ByteString.copyFrom("hello", StandardCharsets.UTF_8))
@@ -111,7 +112,7 @@ public class ProducerServiceTest extends BaseServiceTest {
assertNotNull(e);
assertTrue(e instanceof ExecutionException);
assertTrue(e.getCause() instanceof ProxyException);
assertEquals(Code.NOT_FOUND, ((ProxyException) e.getCause()).getCode());
assertEquals(Code.FORBIDDEN, ((ProxyException) e.getCause()).getCode());
}
}
@@ -163,4 +164,16 @@ public class ProducerServiceTest extends BaseServiceTest {
assertSame(ex, e.getCause());
}
}
// @Test
// public void testForwardMessageToDeadLetterQueue() throws Exception {
// ProducerService producerService = new ProducerService(this.connectorManager);
//
// when(topicRouteCache.getBrokerAddr(anyString())).thenReturn("brokerAddr");
// producerService.forwardMessageToDeadLetterQueue(Context.current(), ForwardMessageToDeadLetterQueueRequest.newBuilder()
// .setMessageId("msgId")
// .setReceiptHandle(createReceiptHandle().encode())
// .set
// .build());
// }
}
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/PullMessageServiceTest.java
+20 -20
View File
@@ -1,17 +1,17 @@
package org.apache.rocketmq.proxy.grpc.v2.service.cluster;
import apache.rocketmq.v1.Broker;
import apache.rocketmq.v1.FilterExpression;
import apache.rocketmq.v1.FilterType;
import apache.rocketmq.v1.Partition;
import apache.rocketmq.v1.PullMessageRequest;
import apache.rocketmq.v1.PullMessageResponse;
import apache.rocketmq.v1.QueryOffsetPolicy;
import apache.rocketmq.v1.QueryOffsetRequest;
import apache.rocketmq.v1.QueryOffsetResponse;
import apache.rocketmq.v1.Resource;
import apache.rocketmq.v2.Broker;
import apache.rocketmq.v2.Code;
import apache.rocketmq.v2.FilterExpression;
import apache.rocketmq.v2.FilterType;
import apache.rocketmq.v2.MessageQueue;
import apache.rocketmq.v2.PullMessageRequest;
import apache.rocketmq.v2.PullMessageResponse;
import apache.rocketmq.v2.QueryOffsetPolicy;
import apache.rocketmq.v2.QueryOffsetRequest;
import apache.rocketmq.v2.QueryOffsetResponse;
import apache.rocketmq.v2.Resource;
import com.google.protobuf.util.Timestamps;
import com.google.rpc.Code;
import io.grpc.Context;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
@@ -49,7 +49,7 @@ public class PullMessageServiceTest extends BaseServiceTest {
when(defaultClient.searchOffset(anyString(), anyString(), anyInt(), anyLong())).thenReturn(CompletableFuture.completedFuture(50L));
QueryOffsetResponse response = pullMessageService.queryOffset(ctx, QueryOffsetRequest.newBuilder()
.setPartition(Partition.newBuilder()
.setMessageQueue(MessageQueue.newBuilder()
.setTopic(Resource.newBuilder()
.setName("topic")
.build())
@@ -58,11 +58,11 @@ public class PullMessageServiceTest extends BaseServiceTest {
.setPolicy(QueryOffsetPolicy.BEGINNING)
.build()
).get();
assertEquals(Code.OK.getNumber(), response.getCommon().getStatus().getCode());
assertEquals(Code.OK, response.getStatus().getCode());
assertEquals(0, response.getOffset());
response = pullMessageService.queryOffset(ctx, QueryOffsetRequest.newBuilder()
.setPartition(Partition.newBuilder()
.setMessageQueue(MessageQueue.newBuilder()
.setTopic(Resource.newBuilder()
.setName("topic")
.build())
@@ -71,11 +71,11 @@ public class PullMessageServiceTest extends BaseServiceTest {
.setPolicy(QueryOffsetPolicy.END)
.build()
).get();
assertEquals(Code.OK.getNumber(), response.getCommon().getStatus().getCode());
assertEquals(Code.OK, response.getStatus().getCode());
assertEquals(100, response.getOffset());
response = pullMessageService.queryOffset(ctx, QueryOffsetRequest.newBuilder()
.setPartition(Partition.newBuilder()
.setMessageQueue(MessageQueue.newBuilder()
.setTopic(Resource.newBuilder()
.setName("topic")
.build())
@@ -85,7 +85,7 @@ public class PullMessageServiceTest extends BaseServiceTest {
.setPolicy(QueryOffsetPolicy.TIME_POINT)
.build()
).get();
assertEquals(Code.OK.getNumber(), response.getCommon().getStatus().getCode());
assertEquals(Code.OK, response.getStatus().getCode());
assertEquals(50, response.getOffset());
}
@@ -109,7 +109,7 @@ public class PullMessageServiceTest extends BaseServiceTest {
Context ctx = Context.current().withDeadlineAfter(3, TimeUnit.SECONDS, Executors.newSingleThreadScheduledExecutor());
PullMessageResponse response = pullMessageService.pullMessage(ctx, PullMessageRequest.newBuilder()
.setPartition(Partition.newBuilder()
.setMessageQueue(MessageQueue.newBuilder()
.setBroker(Broker.newBuilder()
.setName("brokerName")
.build())
@@ -124,8 +124,8 @@ public class PullMessageServiceTest extends BaseServiceTest {
.build())
.get();
assertEquals(Code.OK.getNumber(), response.getCommon().getStatus().getCode());
assertEquals(Code.OK, response.getStatus().getCode());
assertEquals(1, response.getMessagesCount());
assertEquals("msg1", response.getMessages(0).getSystemAttribute().getMessageId());
assertEquals("msg1", response.getMessages(0).getSystemProperties().getMessageId());
}
}
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/RouteServiceTest.java
+44 -21
View File
@@ -20,6 +20,7 @@ package org.apache.rocketmq.proxy.grpc.v2.service.cluster;
import apache.rocketmq.v2.Address;
import apache.rocketmq.v2.AddressScheme;
import apache.rocketmq.v2.Broker;
import apache.rocketmq.v2.ClientSettings;
import apache.rocketmq.v2.Code;
import apache.rocketmq.v2.Endpoints;
import apache.rocketmq.v2.MessageQueue;
@@ -40,13 +41,14 @@ import org.apache.rocketmq.common.constant.PermName;
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.proxy.grpc.v2.adapter.ProxyMode;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.proxy.connector.route.MessageQueueWrapper;
import org.apache.rocketmq.proxy.grpc.v2.adapter.ProxyMode;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.when;
public class RouteServiceTest extends BaseServiceTest {
@@ -146,15 +148,20 @@ public class RouteServiceTest extends BaseServiceTest {
@Test
public void testLocalModeQueryRoute() throws Exception {
RouteService routeService = new RouteService(ProxyMode.LOCAL, this.connectorManager);
CompletableFuture<QueryRouteResponse> future = routeService.queryRoute(Context.current(), QueryRouteRequest.newBuilder()
.setEndpoints(Endpoints.newBuilder()
RouteService routeService = new RouteService(ProxyMode.LOCAL, this.connectorManager, this.grpcClientManager);
ClientSettings clientSettings = ClientSettings.newBuilder()
.setAccessPoint(Endpoints.newBuilder()
.addAddresses(Address.newBuilder()
.setPort(80)
.setHost("host")
.build())
.setScheme(AddressScheme.DOMAIN_NAME)
.build())
.build();
when(grpcClientManager.getClientSettings(anyString())).thenReturn(clientSettings);
CompletableFuture<QueryRouteResponse> future = routeService.queryRoute(Context.current(), QueryRouteRequest.newBuilder()
.setTopic(Resource.newBuilder()
.setName("topic")
.build())
@@ -168,8 +175,9 @@ public class RouteServiceTest extends BaseServiceTest {
@Test
public void testQueryRouteWithInvalidEndpoints() throws Exception {
RouteService routeService = new RouteService(ProxyMode.CLUSTER, this.connectorManager);
RouteService routeService = new RouteService(ProxyMode.CLUSTER, this.connectorManager, this.grpcClientManager);
when(grpcClientManager.getClientSettings(anyString())).thenReturn(ClientSettings.getDefaultInstance());
CompletableFuture<QueryRouteResponse> future = routeService.queryRoute(Context.current(), QueryRouteRequest.newBuilder()
.setTopic(Resource.newBuilder()
.setName("topic")
@@ -182,16 +190,20 @@ public class RouteServiceTest extends BaseServiceTest {
@Test
public void testQueryRoute() throws Exception {
RouteService routeService = new RouteService(ProxyMode.CLUSTER, this.connectorManager);
RouteService routeService = new RouteService(ProxyMode.CLUSTER, this.connectorManager, this.grpcClientManager);
CompletableFuture<QueryRouteResponse> future = routeService.queryRoute(Context.current(), QueryRouteRequest.newBuilder()
.setEndpoints(Endpoints.newBuilder()
ClientSettings clientSettings = ClientSettings.newBuilder()
.setAccessPoint(Endpoints.newBuilder()
.addAddresses(Address.newBuilder()
.setPort(80)
.setHost("host")
.build())
.setScheme(AddressScheme.DOMAIN_NAME)
.build())
.build();
when(grpcClientManager.getClientSettings(anyString())).thenReturn(clientSettings);
CompletableFuture<QueryRouteResponse> future = routeService.queryRoute(Context.current(), QueryRouteRequest.newBuilder()
.setTopic(Resource.newBuilder()
.setName("topic")
.build())
@@ -206,16 +218,20 @@ public class RouteServiceTest extends BaseServiceTest {
@Test
public void testQueryRouteWhenTopicNotExist() throws Exception {
RouteService routeService = new RouteService(ProxyMode.CLUSTER, this.connectorManager);
RouteService routeService = new RouteService(ProxyMode.CLUSTER, this.connectorManager, this.grpcClientManager);
CompletableFuture<QueryRouteResponse> future = routeService.queryRoute(Context.current(), QueryRouteRequest.newBuilder()
.setEndpoints(Endpoints.newBuilder()
ClientSettings clientSettings = ClientSettings.newBuilder()
.setAccessPoint(Endpoints.newBuilder()
.addAddresses(Address.newBuilder()
.setPort(80)
.setHost("host")
.build())
.setScheme(AddressScheme.DOMAIN_NAME)
.build())
.build();
when(grpcClientManager.getClientSettings(anyString())).thenReturn(clientSettings);
CompletableFuture<QueryRouteResponse> future = routeService.queryRoute(Context.current(), QueryRouteRequest.newBuilder()
.setTopic(Resource.newBuilder()
.setName("notExistTopic")
.build())
@@ -227,8 +243,9 @@ public class RouteServiceTest extends BaseServiceTest {
@Test
public void testQueryAssignmentInvalidEndpoints() throws Exception {
RouteService routeService = new RouteService(ProxyMode.CLUSTER, this.connectorManager);
RouteService routeService = new RouteService(ProxyMode.CLUSTER, this.connectorManager, this.grpcClientManager);
when(grpcClientManager.getClientSettings(anyString())).thenReturn(ClientSettings.getDefaultInstance());
CompletableFuture<QueryAssignmentResponse> future = routeService.queryAssignment(Context.current(), QueryAssignmentRequest.newBuilder()
.setTopic(
Resource.newBuilder()
@@ -243,23 +260,26 @@ public class RouteServiceTest extends BaseServiceTest {
@Test
public void testLocalModeQueryAssignment() throws Exception {
RouteService routeService = new RouteService(ProxyMode.LOCAL, this.connectorManager);
RouteService routeService = new RouteService(ProxyMode.LOCAL, this.connectorManager, this.grpcClientManager);
CompletableFuture<QueryAssignmentResponse> future = routeService.queryAssignment(Context.current(), QueryAssignmentRequest.newBuilder()
.setEndpoints(Endpoints.newBuilder()
ClientSettings clientSettings = ClientSettings.newBuilder()
.setAccessPoint(Endpoints.newBuilder()
.addAddresses(Address.newBuilder()
.setPort(80)
.setHost("host")
.build())
.setScheme(AddressScheme.DOMAIN_NAME)
.build())
.build();
when(grpcClientManager.getClientSettings(anyString())).thenReturn(clientSettings);
CompletableFuture<QueryAssignmentResponse> future = routeService.queryAssignment(Context.current(), QueryAssignmentRequest.newBuilder()
.setTopic(Resource.newBuilder()
.setName("topic")
.build())
.setGroup(Resource.newBuilder()
.setName("group")
.build())
.setClientId("clientId")
.build());
QueryAssignmentResponse response = future.get();
@@ -271,23 +291,26 @@ public class RouteServiceTest extends BaseServiceTest {
@Test
public void testQueryAssignment() throws Exception {
RouteService routeService = new RouteService(ProxyMode.CLUSTER, this.connectorManager);
RouteService routeService = new RouteService(ProxyMode.CLUSTER, this.connectorManager, this.grpcClientManager);
CompletableFuture<QueryAssignmentResponse> future = routeService.queryAssignment(Context.current(), QueryAssignmentRequest.newBuilder()
.setEndpoints(Endpoints.newBuilder()
ClientSettings clientSettings = ClientSettings.newBuilder()
.setAccessPoint(Endpoints.newBuilder()
.addAddresses(Address.newBuilder()
.setPort(80)
.setHost("host")
.build())
.setScheme(AddressScheme.DOMAIN_NAME)
.build())
.build();
when(grpcClientManager.getClientSettings(anyString())).thenReturn(clientSettings);
CompletableFuture<QueryAssignmentResponse> future = routeService.queryAssignment(Context.current(), QueryAssignmentRequest.newBuilder()
.setTopic(Resource.newBuilder()
.setName("topic")
.build())
.setGroup(Resource.newBuilder()
.setName("group")
.build())
.setClientId("clientId")
.build());
QueryAssignmentResponse response = future.get();
@@ -297,4 +320,4 @@ public class RouteServiceTest extends BaseServiceTest {
assertEquals("host", response.getAssignments(0).getMessageQueue().getBroker().getEndpoints().getAddresses(0).getHost());
}
}
}
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/TransactionServiceTest.java
+8 -12
View File
@@ -1,17 +1,16 @@
package org.apache.rocketmq.proxy.grpc.v2.service.cluster;
import apache.rocketmq.v1.EndTransactionRequest;
import apache.rocketmq.v1.EndTransactionResponse;
import apache.rocketmq.v1.PollCommandResponse;
import apache.rocketmq.v1.Resource;
import com.google.rpc.Code;
import apache.rocketmq.v2.Code;
import apache.rocketmq.v2.EndTransactionRequest;
import apache.rocketmq.v2.EndTransactionResponse;
import apache.rocketmq.v2.TelemetryCommand;
import io.grpc.Context;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.rocketmq.common.protocol.header.EndTransactionRequestHeader;
import org.apache.rocketmq.proxy.channel.ChannelManager;
import org.apache.rocketmq.proxy.connector.transaction.TransactionId;
import org.apache.rocketmq.proxy.connector.transaction.TransactionStateCheckRequest;
import org.apache.rocketmq.proxy.grpc.v1.adapter.channel.GrpcClientChannel;
import org.apache.rocketmq.proxy.grpc.v2.adapter.channel.GrpcClientChannel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.assertj.core.util.Lists;
import org.junit.Test;
@@ -60,8 +59,8 @@ public class TransactionServiceTest extends BaseServiceTest {
createMessageExt("msgId", "msgId")
));
assertTrue(writeDataRef.get() instanceof PollCommandResponse);
PollCommandResponse response = (PollCommandResponse) writeDataRef.get();
assertTrue(writeDataRef.get() instanceof TelemetryCommand);
TelemetryCommand response = (TelemetryCommand) writeDataRef.get();
assertEquals(transactionId.getProxyTransactionId(), response.getRecoverOrphanedTransactionCommand().getTransactionId());
}
@@ -79,14 +78,11 @@ public class TransactionServiceTest extends BaseServiceTest {
}).when(producerClient).endTransaction(anyString(), any());
EndTransactionResponse response = transactionService.endTransaction(Context.current(), EndTransactionRequest.newBuilder()
.setGroup(Resource.newBuilder()
.setName("group")
.build())
.setTransactionId(transactionId.getProxyTransactionId())
.build()
).get();
assertEquals(Code.OK.getNumber(), response.getCommon().getStatus().getCode());
assertEquals(Code.OK, response.getStatus().getCode());
assertEquals(transactionId.getBrokerTransactionId(), headerRef.get().getTransactionId());
assertEquals("127.0.0.1:8080", brokerAddrRef.get());
}