This is an automated email from the ASF dual-hosted git repository.
yuz10 pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 37af1437e6 [ISSUE #10478] Support Lite messages in proxy local mode
(#11053)
37af1437e6 is described below
commit 37af1437e60e48009105c4a32851fb67f4c7acb8
Author: Rui Yin <[email protected]>
AuthorDate: Sat Oct 3 15:27:10 2026 +0800
[ISSUE #10478] Support Lite messages in proxy local mode (#11053)
* [ISSUE #10478] Support Lite messages in proxy local mode
* [ISSUE #10478] Cover Lite pop edge cases and log invalid metadata
---
.../proxy/service/LocalServiceManager.java | 13 +-
.../proxy/service/message/LocalMessageService.java | 66 +++++++-
.../service/message/LocalMessageServiceTest.java | 171 +++++++++++++++++++++
3 files changed, 248 insertions(+), 2 deletions(-)
diff --git
a/proxy/src/main/java/org/apache/rocketmq/proxy/service/LocalServiceManager.java
b/proxy/src/main/java/org/apache/rocketmq/proxy/service/LocalServiceManager.java
index 6e7827c4eb..111ce0fd52 100644
---
a/proxy/src/main/java/org/apache/rocketmq/proxy/service/LocalServiceManager.java
+++
b/proxy/src/main/java/org/apache/rocketmq/proxy/service/LocalServiceManager.java
@@ -55,8 +55,10 @@ public class LocalServiceManager extends
AbstractStartAndShutdown implements Ser
private final ProxyRelayService proxyRelayService;
private final MetadataService metadataService;
private final AdminService adminService;
+ private final LiteSubscriptionService liteSubscriptionService;
private final MQClientAPIFactory mqClientAPIFactory;
+ private final MQClientAPIFactory liteSubscriptionAPIFactory;
private final ChannelManager channelManager;
private final ScheduledExecutorService scheduledExecutorService =
ThreadUtils.newSingleThreadScheduledExecutor(
@@ -82,11 +84,20 @@ public class LocalServiceManager extends
AbstractStartAndShutdown implements Ser
this.proxyRelayService = new LocalProxyRelayService(brokerController,
this.transactionService);
this.metadataService = new LocalMetadataService(brokerController);
this.adminService = new DefaultAdminService(this.mqClientAPIFactory);
+ this.liteSubscriptionAPIFactory = new MQClientAPIFactory(
+ nameserverAccessConfig,
+ "LiteSubscription_",
+ 1,
+ new DoNothingClientRemotingProcessor(null),
+ rpcHook,
+ scheduledExecutorService);
+ this.liteSubscriptionService = new
LiteSubscriptionService(this.topicRouteService,
this.liteSubscriptionAPIFactory);
this.init();
}
protected void init() {
this.appendStartAndShutdown(this.mqClientAPIFactory);
+ this.appendStartAndShutdown(this.liteSubscriptionAPIFactory);
this.appendStartAndShutdown(this.topicRouteService);
this.appendStartAndShutdown(this.adminService);
this.appendStartAndShutdown(new LocalServiceManagerStartAndShutdown());
@@ -134,7 +145,7 @@ public class LocalServiceManager extends
AbstractStartAndShutdown implements Ser
@Override
public LiteSubscriptionService getLiteSubscriptionService() {
- return null;
+ return liteSubscriptionService;
}
private class LocalServiceManagerStartAndShutdown implements
StartAndShutdown {
diff --git
a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java
b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java
index c93fa93983..a14b09665e 100644
---
a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java
+++
b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java
@@ -27,6 +27,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import org.apache.commons.lang3.NotImplementedException;
+import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.client.consumer.AckResult;
import org.apache.rocketmq.client.consumer.AckStatus;
@@ -36,6 +37,7 @@ import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.consumer.ReceiptHandle;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageBatch;
@@ -68,6 +70,7 @@ import
org.apache.rocketmq.remoting.protocol.header.ExtraInfoUtil;
import org.apache.rocketmq.remoting.protocol.header.GetMaxOffsetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.GetMinOffsetRequestHeader;
import
org.apache.rocketmq.remoting.protocol.header.PopLiteMessageRequestHeader;
+import
org.apache.rocketmq.remoting.protocol.header.PopLiteMessageResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.PopMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.PopMessageResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.PullMessageRequestHeader;
@@ -200,7 +203,68 @@ public class LocalMessageService implements MessageService
{
@Override
public CompletableFuture<PopResult> popLiteMessage(ProxyContext ctx,
AddressableMessageQueue messageQueue,
PopLiteMessageRequestHeader requestHeader, long timeoutMillis) {
- throw new NotImplementedException();
+ RemotingCommand request =
LocalRemotingCommand.createRequestCommand(RequestCode.POP_LITE_MESSAGE,
requestHeader, ctx.getLanguage());
+ CompletableFuture<RemotingCommand> future = new CompletableFuture<>();
+ SimpleChannel channel = channelManager.createInvocationChannel(ctx);
+ InvocationContext invocationContext = new InvocationContext(future);
+ channel.registerInvocationContext(request.getOpaque(),
invocationContext);
+ ChannelHandlerContext channelHandlerContext =
channel.getChannelHandlerContext();
+ try {
+ RemotingCommand response =
brokerController.getPopLiteMessageProcessor().processRequest(channelHandlerContext,
request);
+ if (response != null) {
+ invocationContext.handle(response);
+ channel.eraseInvocationContext(request.getOpaque());
+ }
+ } catch (Exception e) {
+ future.completeExceptionally(e);
+ channel.eraseInvocationContext(request.getOpaque());
+ log.error("Failed to process popLiteMessage command", e);
+ }
+ return future.thenApply(r -> {
+ PopStatus popStatus;
+ List<MessageExt> messageExtList = new ArrayList<>();
+ switch (r.getCode()) {
+ case ResponseCode.SUCCESS:
+ popStatus = PopStatus.FOUND;
+ messageExtList =
MessageDecoder.decodesBatch(ByteBuffer.wrap(r.getBody()), true, false, true);
+ break;
+ case ResponseCode.POLLING_FULL:
+ popStatus = PopStatus.POLLING_FULL;
+ break;
+ case ResponseCode.POLLING_TIMEOUT:
+ case ResponseCode.PULL_NOT_FOUND:
+ popStatus = PopStatus.POLLING_NOT_FOUND;
+ break;
+ default:
+ throw new
ProxyException(ProxyExceptionCode.INTERNAL_SERVER_ERROR, r.getRemark());
+ }
+ PopResult popResult = new PopResult(popStatus, messageExtList);
+ if (popStatus != PopStatus.FOUND) {
+ return popResult;
+ }
+ PopLiteMessageResponseHeader responseHeader =
(PopLiteMessageResponseHeader) r.readCustomHeader();
+ List<Integer> orderCountList =
ExtraInfoUtil.parseLiteOrderCountInfo(responseHeader.getOrderCountInfo(),
messageExtList.size());
+ for (int i = 0; i < messageExtList.size(); i++) {
+ MessageExt messageExt = messageExtList.get(i);
+ String[] queues =
StringUtils.split(messageExt.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH),
MixAll.LMQ_DISPATCH_SEPARATOR);
+ String[] queueOffsets =
StringUtils.split(messageExt.getProperty(MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET),
MixAll.LMQ_DISPATCH_SEPARATOR);
+ if (queues == null || queueOffsets == null || queues.length !=
1 || queues.length != queueOffsets.length) {
+ log.warn("Invalid Lite dispatch metadata, messageId={},
brokerName={}",
+ messageExt.getMsgId(), messageQueue.getBrokerName());
+ continue;
+ }
+ messageExt.getProperties().put(MessageConst.PROPERTY_POP_CK,
+ ExtraInfoUtil.buildExtraInfo(0,
responseHeader.getPopTime(), responseHeader.getInvisibleTime(),
+ responseHeader.getReviveQid(),
messageQueue.getTopic(), messageQueue.getBrokerName(), 0,
+ Long.parseLong(queueOffsets[0])));
+
messageExt.getProperties().computeIfAbsent(MessageConst.PROPERTY_FIRST_POP_TIME,
+ k -> String.valueOf(responseHeader.getPopTime()));
+ messageExt.setBrokerName(messageQueue.getBrokerName());
+ messageExt.setReconsumeTimes(orderCountList != null ?
orderCountList.get(i) : 0);
+ messageExt.setQueueOffset(Long.parseLong(queueOffsets[0]));
+ }
+ return popResult;
+ });
}
@Override
diff --git
a/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java
b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java
index 52ba521f80..a0fe3b9c98 100644
---
a/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java
+++
b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java
@@ -28,10 +28,12 @@ import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.processor.AckMessageProcessor;
import org.apache.rocketmq.broker.processor.ChangeInvisibleTimeProcessor;
import org.apache.rocketmq.broker.processor.EndTransactionProcessor;
+import org.apache.rocketmq.broker.processor.PopLiteMessageProcessor;
import org.apache.rocketmq.broker.processor.PopMessageProcessor;
import org.apache.rocketmq.broker.processor.RecallMessageProcessor;
import org.apache.rocketmq.broker.processor.SendMessageProcessor;
@@ -46,6 +48,7 @@ import org.apache.rocketmq.common.consumer.ReceiptHandle;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageBatch;
import org.apache.rocketmq.common.message.MessageClientIDSetter;
+import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
@@ -70,6 +73,8 @@ import
org.apache.rocketmq.remoting.protocol.header.EndTransactionRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ExtraInfoUtil;
import org.apache.rocketmq.remoting.protocol.header.PopMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.PopMessageResponseHeader;
+import
org.apache.rocketmq.remoting.protocol.header.PopLiteMessageRequestHeader;
+import
org.apache.rocketmq.remoting.protocol.header.PopLiteMessageResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.RecallMessageRequestHeader;
import
org.apache.rocketmq.remoting.protocol.header.RecallMessageResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader;
@@ -95,6 +100,8 @@ public class LocalMessageServiceTest extends InitConfigTest {
@Mock
private PopMessageProcessor popMessageProcessorMock;
@Mock
+ private PopLiteMessageProcessor popLiteMessageProcessorMock;
+ @Mock
private ChangeInvisibleTimeProcessor changeInvisibleTimeProcessorMock;
@Mock
private AckMessageProcessor ackMessageProcessorMock;
@@ -126,6 +133,7 @@ public class LocalMessageServiceTest extends InitConfigTest
{
channelManager = new ChannelManager();
Mockito.when(brokerControllerMock.getSendMessageProcessor()).thenReturn(sendMessageProcessorMock);
Mockito.when(brokerControllerMock.getPopMessageProcessor()).thenReturn(popMessageProcessorMock);
+
Mockito.when(brokerControllerMock.getPopLiteMessageProcessor()).thenReturn(popLiteMessageProcessorMock);
Mockito.when(brokerControllerMock.getChangeInvisibleTimeProcessor()).thenReturn(changeInvisibleTimeProcessorMock);
Mockito.when(brokerControllerMock.getAckMessageProcessor()).thenReturn(ackMessageProcessorMock);
Mockito.when(brokerControllerMock.getEndTransactionProcessor()).thenReturn(endTransactionProcessorMock);
@@ -277,6 +285,169 @@ public class LocalMessageServiceTest extends
InitConfigTest {
}));
}
+ @Test
+ public void testPopLiteMessageWriteAndFlush() throws Exception {
+ long popTime = System.currentTimeMillis();
+ long invisibleTime = 3000L;
+ List<MessageExt> messages = new ArrayList<>();
+ MessageExt message1 = buildMessageExt(topic, 0, 100L);
+
message1.getProperties().put(MessageConst.PROPERTY_INNER_MULTI_DISPATCH,
"%LMQ%$topic$lite");
+
message1.getProperties().put(MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET,
"0");
+ messages.add(message1);
+ MessageExt message2 = buildMessageExt(topic, 0, 101L);
+
message2.getProperties().put(MessageConst.PROPERTY_INNER_MULTI_DISPATCH,
"%LMQ%$topic$lite");
+
message2.getProperties().put(MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET,
"1");
+ messages.add(message2);
+ byte[] body = ByteBuffer.allocate(MessageDecoder.encode(message1,
false).length + MessageDecoder.encode(message2, false).length)
+ .put(MessageDecoder.encode(message1,
false)).put(MessageDecoder.encode(message2, false)).array();
+ PopLiteMessageRequestHeader requestHeader = new
PopLiteMessageRequestHeader();
+ requestHeader.setInvisibleTime(invisibleTime);
+
Mockito.when(popLiteMessageProcessorMock.processRequest(Mockito.any(SimpleChannelHandlerContext.class),
Mockito.argThat(argument ->
+ argument.getCode() == RequestCode.POP_LITE_MESSAGE
+ && argument.readCustomHeader() instanceof
PopLiteMessageRequestHeader))).thenAnswer(invocation -> {
+ SimpleChannelHandlerContext channelHandlerContext =
invocation.getArgument(0);
+ RemotingCommand request = invocation.getArgument(1);
+ RemotingCommand response =
RemotingCommand.createResponseCommand(PopLiteMessageResponseHeader.class);
+ response.setOpaque(request.getOpaque());
+ response.setCode(ResponseCode.SUCCESS);
+ response.setBody(body);
+ PopLiteMessageResponseHeader responseHeader =
(PopLiteMessageResponseHeader) response.readCustomHeader();
+ responseHeader.setPopTime(popTime);
+ responseHeader.setInvisibleTime(invisibleTime);
+ responseHeader.setReviveQid(1);
+ channelHandlerContext.writeAndFlush(response);
+ return null;
+ });
+
+ PopResult result = localMessageService.popLiteMessage(proxyContext,
+ new AddressableMessageQueue(new MessageQueue(topic, brokerName,
queueId), ""), requestHeader, 1000L).get();
+ assertThat(result.getPopStatus()).isEqualTo(PopStatus.FOUND);
+ assertThat(result.getMsgFoundList()).hasSize(2);
+
assertThat(result.getMsgFoundList().get(0).getQueueOffset()).isEqualTo(0L);
+
assertThat(result.getMsgFoundList().get(1).getQueueOffset()).isEqualTo(1L);
+
assertThat(result.getMsgFoundList().get(0).getBrokerName()).isEqualTo(brokerName);
+ }
+
+ @Test
+ public void testPopLiteMessagePollingResponses() throws Exception {
+ int[] codes = {ResponseCode.POLLING_FULL,
ResponseCode.POLLING_TIMEOUT, ResponseCode.PULL_NOT_FOUND};
+ PopStatus[] statuses = {PopStatus.POLLING_FULL,
PopStatus.POLLING_NOT_FOUND, PopStatus.POLLING_NOT_FOUND};
+ for (int i = 0; i < codes.length; i++) {
+ PopResult result =
invokeLiteResponse(RemotingCommand.createResponseCommand(codes[i], ""));
+ assertThat(result.getPopStatus()).isEqualTo(statuses[i]);
+ assertThat(result.getMsgFoundList()).isEmpty();
+ }
+ }
+
+ @Test
+ public void testPopLiteMessageEmptyBatch() throws Exception {
+ PopResult result = invokeLiteResponse(buildLiteResponse(null));
+ assertThat(result.getPopStatus()).isEqualTo(PopStatus.FOUND);
+ assertThat(result.getMsgFoundList()).isEmpty();
+ }
+
+ @Test
+ public void testPopLiteMessageMalformedDispatchProperties() throws
Exception {
+ String[][] properties = {
+ {null, "7"}, {"%LMQ%$topic$lite", null},
+ {"%LMQ%$topic$lite,%LMQ%$topic$other", "7,8"},
{"%LMQ%$topic$lite", "7,8"}
+ };
+ for (String[] property : properties) {
+ MessageExt message = buildMessageExt(topic, 0, 100L);
+ if (property[0] != null) {
+
message.getProperties().put(MessageConst.PROPERTY_INNER_MULTI_DISPATCH,
property[0]);
+ }
+ if (property[1] != null) {
+
message.getProperties().put(MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET,
property[1]);
+ }
+ PopResult result = invokeLiteResponse(buildLiteResponse(null,
message));
+ assertThat(result.getMsgFoundList()).hasSize(1);
+
assertThat(result.getMsgFoundList().get(0).getQueueOffset()).isEqualTo(100L);
+
assertThat(result.getMsgFoundList().get(0).getProperty(MessageConst.PROPERTY_POP_CK)).isNull();
+ }
+ }
+
+ @Test
+ public void testPopLiteMessageMissingOrMismatchedOrderCounts() throws
Exception {
+ for (String counts : new String[] {null, "", "3", "3;4;5"}) {
+ PopResult result = invokeLiteResponse(buildLiteResponse(counts,
buildLiteMessage(7L), buildLiteMessage(8L)));
+ assertThat(result.getMsgFoundList()).hasSize(2);
+ for (MessageExt message : result.getMsgFoundList()) {
+ assertThat(message.getReconsumeTimes()).isZero();
+
assertThat(message.getProperty(MessageConst.PROPERTY_POP_CK)).isNotNull();
+ }
+ }
+ }
+
+ @Test
+ public void testPopLiteMessageOrderCountsAndReceiptMetadata() throws
Exception {
+ PopResult result = invokeLiteResponse(buildLiteResponse("3;4",
buildLiteMessage(7L), buildLiteMessage(8L)));
+ for (int i = 0; i < 2; i++) {
+ MessageExt message = result.getMsgFoundList().get(i);
+ assertThat(message.getReconsumeTimes()).isEqualTo(i + 3);
+ assertThat(message.getQueueOffset()).isEqualTo(i + 7L);
+ assertThat(message.getBrokerName()).isEqualTo(brokerName);
+
assertThat(message.getProperty(MessageConst.PROPERTY_FIRST_POP_TIME)).isEqualTo("123456");
+
assertThat(message.getProperty(MessageConst.PROPERTY_POP_CK)).isEqualTo(
+ ExtraInfoUtil.buildExtraInfo(0, 123456L, 3000L, 1, topic,
brokerName, 0, i + 7L));
+ }
+ }
+
+ @Test
+ public void testPopLiteMessageBrokerError() throws Exception {
+ ExecutionException exception = catchThrowableOfType(() ->
invokeLiteResponse(
+ RemotingCommand.createResponseCommand(ResponseCode.SYSTEM_ERROR,
"lite failure")), ExecutionException.class);
+ assertThat(exception.getCause()).isInstanceOf(ProxyException.class);
+ assertThat(((ProxyException)
exception.getCause()).getCode()).isEqualTo(ProxyExceptionCode.INTERNAL_SERVER_ERROR);
+ assertThat(exception.getCause()).hasMessage("lite failure");
+ }
+
+ @Test
+ public void testPopLiteMessageProcessorException() throws Exception {
+ RemotingCommandException failure = new RemotingCommandException("lite
failure");
+
Mockito.when(popLiteMessageProcessorMock.processRequest(Mockito.any(SimpleChannelHandlerContext.class),
Mockito.any()))
+ .thenThrow(failure);
+ ExecutionException exception = catchThrowableOfType(() ->
localMessageService.popLiteMessage(proxyContext,
+ null, new PopLiteMessageRequestHeader(), 1000L).get(5,
TimeUnit.SECONDS), ExecutionException.class);
+ assertThat(exception.getCause()).isSameAs(failure);
+ }
+
+ private MessageExt buildLiteMessage(long offset) {
+ MessageExt message = buildMessageExt(topic, 0, 100L);
+
message.getProperties().put(MessageConst.PROPERTY_INNER_MULTI_DISPATCH,
"%LMQ%$topic$lite");
+
message.getProperties().put(MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET,
Long.toString(offset));
+ return message;
+ }
+
+ private RemotingCommand buildLiteResponse(String orderCounts,
MessageExt... messages) throws Exception {
+ List<byte[]> encoded = new ArrayList<>();
+ int size = 0;
+ for (MessageExt message : messages) {
+ byte[] bytes = MessageDecoder.encode(message, false);
+ encoded.add(bytes);
+ size += bytes.length;
+ }
+ ByteBuffer body = ByteBuffer.allocate(size);
+ encoded.forEach(body::put);
+ RemotingCommand response =
RemotingCommand.createResponseCommand(PopLiteMessageResponseHeader.class);
+ response.setCode(ResponseCode.SUCCESS);
+ response.setBody(body.array());
+ PopLiteMessageResponseHeader header = (PopLiteMessageResponseHeader)
response.readCustomHeader();
+ header.setPopTime(123456L);
+ header.setInvisibleTime(3000L);
+ header.setReviveQid(1);
+ header.setOrderCountInfo(orderCounts);
+ return response;
+ }
+
+ private PopResult invokeLiteResponse(RemotingCommand response) throws
Exception {
+
Mockito.when(popLiteMessageProcessorMock.processRequest(Mockito.any(SimpleChannelHandlerContext.class),
Mockito.any()))
+ .thenReturn(response);
+ return localMessageService.popLiteMessage(proxyContext,
+ new AddressableMessageQueue(new MessageQueue(topic, brokerName,
queueId), ""),
+ new PopLiteMessageRequestHeader(), 1000L).get(5, TimeUnit.SECONDS);
+ }
+
@Test
public void testPopMessageWriteAndFlush() throws Exception {
int reviveQueueId = 1;
