From edaffe69060bfd95f63e291308d28ce03738a1fb Mon Sep 17 00:00:00 2001 From: lizhimins <707364882@qq.com> Date: Fri, 12 Sep 2025 17:26:16 +0800 Subject: [PATCH] [ISSUE #9695] Not use pull offset when use pop orderly consume (#9696) --- .../rocketmq/broker/pop/PopConsumerService.java | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java index 1138ff4afe..277a4999cf 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java @@ -199,8 +199,14 @@ public class PopConsumerService extends ServiceThread { return context; } - public long getPopOffset(String groupId, String topicId, int queueId, int initMode) { - long offset = this.brokerController.getConsumerOffsetManager().queryPullOffset(groupId, topicId, queueId); + public long getPopOffset(String groupId, String topicId, int queueId, int initMode, boolean fifo) { + + // For FIFO messages, the pull offset is not used. + // This preserves compatibility when switching from pull consumer to pop consumer. + long offset = fifo ? + this.brokerController.getConsumerOffsetManager().queryOffset(groupId, topicId, queueId) : + this.brokerController.getConsumerOffsetManager().queryPullOffset(groupId, topicId, queueId); + if (offset < 0L) { try { offset = this.brokerController.getPopMessageProcessor() @@ -309,7 +315,7 @@ public class PopConsumerService extends ServiceThread { result.addRestCount(this.getPendingFilterCount(groupId, topicId, queueId)); return CompletableFuture.completedFuture(result); } else { - final long consumeOffset = this.getPopOffset(groupId, topicId, queueId, result.getInitMode()); + final long consumeOffset = this.getPopOffset(groupId, topicId, queueId, result.getInitMode(), result.isFifo()); return getMessageAsync(clientHost, groupId, topicId, queueId, consumeOffset, remain, filter) .thenApply(getMessageResult -> handleGetMessageResult( result, getMessageResult, topicId, queueId, retryType, consumeOffset));