opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-30 18:10:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #9695] Not use pull offset when use pop orderly consume (#9696)

Browse Source
This commit is contained in:
lizhimins
2025-09-12 17:26:16 +08:00
committed by GitHub
parent 7967edfc5e
commit edaffe6906
1 changed files with 9 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java
+9 -3
View File
@@ -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));