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 #9339] Fix pop update consumption offset when message are filtered (#9340)

Browse Source
This commit is contained in:
lizhimins
2025-04-16 16:00:37 +08:00
committed by GitHub
parent 9adff6a39e
commit 854369aa38
4 changed files with 17 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java
+12 -11
View File
@@ -162,17 +162,15 @@ public class PopConsumerService extends ServiceThread {
return result;
}
public PopConsumerContext addGetMessageResult(PopConsumerContext context, GetMessageResult result,
public PopConsumerContext handleGetMessageResult(PopConsumerContext context, GetMessageResult result,
String topicId, int queueId, PopConsumerRecord.RetryType retryType, long offset) {
if (result.getStatus() == GetMessageStatus.FOUND && !result.getMessageQueueOffset().isEmpty()) {
if (GetMessageStatus.FOUND.equals(result.getStatus()) && !result.getMessageQueueOffset().isEmpty()) {
if (context.isFifo()) {
this.setFifoBlocked(context, context.getGroupId(), topicId, queueId, result.getMessageQueueOffset());
}
// build request header here
// build response header here
context.addGetMessageResult(result, topicId, queueId, retryType, offset);
if (brokerConfig.isPopConsumerKVServiceLog()) {
log.info("PopConsumerService pop, time={}, invisible={}, " +
"groupId={}, topic={}, queueId={}, offset={}, attemptId={}",
@@ -181,20 +179,23 @@ public class PopConsumerService extends ServiceThread {
}
}
if (!context.isFifo() && result.getNextBeginOffset() > OFFSET_NOT_EXIST) {
long commitOffset = offset;
if (context.isFifo()) {
if (!GetMessageStatus.FOUND.equals(result.getStatus())) {
commitOffset = result.getNextBeginOffset();
}
} else {
this.brokerController.getConsumerOffsetManager().commitPullOffset(
context.getClientHost(), context.getGroupId(), topicId, queueId, result.getNextBeginOffset());
long commitOffset = result.getStatus() == GetMessageStatus.FOUND ? offset : result.getNextBeginOffset();
if (brokerConfig.isEnablePopBufferMerge() && popConsumerCache != null) {
long minOffset = popConsumerCache.getMinOffsetInCache(context.getGroupId(), topicId, queueId);
if (minOffset != OFFSET_NOT_EXIST) {
commitOffset = minOffset;
}
}
this.brokerController.getConsumerOffsetManager().commitOffset(
context.getClientHost(), context.getGroupId(), topicId, queueId, commitOffset);
}
this.brokerController.getConsumerOffsetManager().commitOffset(
context.getClientHost(), context.getGroupId(), topicId, queueId, commitOffset);
return context;
}
@@ -310,7 +311,7 @@ public class PopConsumerService extends ServiceThread {
} else {
final long consumeOffset = this.getPopOffset(groupId, topicId, queueId, result.getInitMode());
return getMessageAsync(clientHost, groupId, topicId, queueId, consumeOffset, remain, filter)
.thenApply(getMessageResult -> addGetMessageResult(
.thenApply(getMessageResult -> handleGetMessageResult(
result, getMessageResult, topicId, queueId, retryType, consumeOffset));
}
});
broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java
+3 -1
View File
@@ -680,7 +680,9 @@ public class PopMessageProcessor implements NettyRequestProcessor {
CompletableFuture<Long> future = new CompletableFuture<>();
if (!queueLockManager.tryLock(lockKey)) {
try {
restNum = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId) - offset + restNum;
if (!requestHeader.isOrder()) {
restNum = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId) - offset + restNum;
}
future.complete(restNum);
} catch (ConsumeQueueException e) {
future.completeExceptionally(e);
broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java
+1 -1
View File
@@ -195,7 +195,7 @@ public class PopConsumerServiceTest {
GetMessageResult result = new GetMessageResult();
result.setStatus(GetMessageStatus.FOUND);
result.getMessageQueueOffset().add(100L);
consumerService.addGetMessageResult(
consumerService.handleGetMessageResult(
context, result, topicId, queueId, PopConsumerRecord.RetryType.NORMAL_TOPIC, 100);
Assert.assertEquals(1, context.getGetMessageResultList().size());
}
store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+1 -1
View File
@@ -875,7 +875,7 @@ public class DefaultMessageStore implements MessageStore {
boolean isInMem = estimateInMemByCommitOffset(offsetPy, maxOffsetPy);
if ((cqUnit.getQueueOffset() - offset) * consumeQueue.getUnitSize() > maxFilterMessageSize) {
if ((cqUnit.getQueueOffset() - offset) * consumeQueue.getUnitSize() >= maxFilterMessageSize) {
break;
}