opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-01 15:47:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #9369] Fix reset offset commit pull offset when use pop consumer service (#9370)

Browse Source
This commit is contained in:
lizhimins
2025-04-28 14:02:22 +08:00
committed by GitHub
parent 9f10d38a23
commit 3c2e81dcd4
4 changed files with 9 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
-4
View File
@@ -1555,10 +1555,6 @@ public class BrokerController {
this.consumerFilterManager.persist();
}
if (this.consumerOrderInfoManager != null) {
this.consumerOrderInfoManager.persist();
}
if (this.scheduleMessageService != null) {
this.scheduleMessageService.persist();
this.scheduleMessageService.shutdown();
broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java
+4
View File
@@ -283,6 +283,10 @@ public class ConsumerOffsetManager extends ConfigManager {
return offset;
}
public void clearPullOffset(final String group, final String topic) {
this.pullOffsetTable.remove(topic + TOPIC_GROUP_SEPARATOR + group);
}
@Override
public String encode() {
return this.encode(false);
broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
+1 -2
View File
@@ -2263,8 +2263,7 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
}
if (brokerController.getBrokerConfig().isPopConsumerKVServiceEnable()) {
brokerController.getPopConsumerService().clearCache(group, topic, entry.getKey());
brokerController.getConsumerOffsetManager().commitPullOffset(
"ResetOffsetInner", group, topic, entry.getKey(), entry.getValue());
brokerController.getConsumerOffsetManager().clearPullOffset(group, topic);
}
body.getOffsetTable().put(new MessageQueue(topic, brokerName, entry.getKey()), entry.getValue());
}
broker/src/test/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManagerTest.java
+4
View File
@@ -77,6 +77,10 @@ public class ConsumerOffsetManagerTest {
consumerOffsetManager.commitPullOffset("Pull", group, topic, 0, 100);
consumerOffsetManager.removeOffset(group);
Assert.assertFalse(consumerOffsetManager.getOffsetTable().containsKey(topic + TOPIC_GROUP_SEPARATOR + group));
consumerOffsetManager.commitPullOffset("Pull", group, topic, 0, 100);
consumerOffsetManager.clearPullOffset(group, topic);
Assert.assertEquals(-1L, consumerOffsetManager.queryPullOffset(group, topic, 0));
}
@Test