diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/BaseServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/BaseServiceTest.java index f7d0e9a98d..347a340ae0 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/BaseServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/BaseServiceTest.java @@ -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; + } } diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ConsumerServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ConsumerServiceTest.java index 0a8c76a3a5..ccfb2a0fa0 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ConsumerServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ConsumerServiceTest.java @@ -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(); + } } \ No newline at end of file diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultReadQueueSelectorTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultReadQueueSelectorTest.java index 167ff99027..823b7437fd 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultReadQueueSelectorTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultReadQueueSelectorTest.java @@ -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()) diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultWriteQueueSelectorTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultWriteQueueSelectorTest.java index c7aff6cd7f..b68d935530 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultWriteQueueSelectorTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultWriteQueueSelectorTest.java @@ -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()); diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ForwardClientServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ForwardClientServiceTest.java index c2d9fdedea..2c0aa30309 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ForwardClientServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ForwardClientServiceTest.java @@ -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 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()); diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ProducerServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ProducerServiceTest.java index 944ecce038..1b4c15fd52 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ProducerServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ProducerServiceTest.java @@ -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 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()); +// } } \ No newline at end of file diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/PullMessageServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/PullMessageServiceTest.java index b5cee9bbb9..23060b1b86 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/PullMessageServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/PullMessageServiceTest.java @@ -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()); } } \ No newline at end of file diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/RouteServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/RouteServiceTest.java index 395f30b0c7..be88441fa3 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/RouteServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/RouteServiceTest.java @@ -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 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 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 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 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 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 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 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 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 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 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 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 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()); } -} +} diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/TransactionServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/TransactionServiceTest.java index cb108e7e83..5909ffac16 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/TransactionServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/TransactionServiceTest.java @@ -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()); }