This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12417-d710e0678d0357ac3f22ffb694f32cb21da36ac6 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 6934fd00626e6347f86320b4f97d7c4cddc0fb59 Author: Sepuri Sai Krishna <[email protected]> AuthorDate: Mon Sep 21 15:36:33 2026 +0000 [Improve][Connector-V2] Note the multi-topic early return in currentOffsets (#12417) --- .../rocketmq/common/RocketMqAdminUtil.java | 10 +++++ .../rocketmq/common/RocketMqAdminUtilTest.java | 43 ++++++++++++++++++++++ 2 files changed, 53 insertions(+) diff --git a/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/common/RocketMqAdminUtil.java b/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/common/RocketMqAdminUtil.java index 589d917cbc..df164fe0d2 100644 --- a/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/common/RocketMqAdminUtil.java +++ b/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/common/RocketMqAdminUtil.java @@ -339,6 +339,16 @@ public class RocketMqAdminUtil { && topicRouteAvailable(adminClient, topic)) { // The retry topic is per group, not per topic, so this applies to every topic // in the request and the whole lookup is legitimately empty. + // + // Returning here discards anything consumerOffsets has already collected for + // earlier topics, and topics is a supported multi-topic list + // (RocketMqSourceOptions.TOPICS). That is safe only under the invariant + // above: a group cannot commit an offset for any topic without first + // registering, and registering is what creates the retry topic, so a missing + // retry topic means no topic in the list has committed anything and the map + // is necessarily still empty here. If that ever stops holding, this has to + // become a continue that keeps the earlier offsets, otherwise a later + // cold-start topic silently rewinds the topics already read. log.warn( "Consumer group {} has no retry topic yet, so it has never registered " + "and has committed nothing. Topic {} still resolves, so this " diff --git a/seatunnel-connectors-v2/connector-rocketmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/common/RocketMqAdminUtilTest.java b/seatunnel-connectors-v2/connector-rocketmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/common/RocketMqAdminUtilTest.java index daae2bbcc7..6072ddc389 100644 --- a/seatunnel-connectors-v2/connector-rocketmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/common/RocketMqAdminUtilTest.java +++ b/seatunnel-connectors-v2/connector-rocketmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/common/RocketMqAdminUtilTest.java @@ -31,8 +31,10 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.mockito.Mockito; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.Map; /** @@ -50,6 +52,7 @@ class RocketMqAdminUtilTest { private static final String GROUP = "test-group"; private static final String TOPIC = "test-topic"; + private static final String OTHER_TOPIC = "other-test-topic"; /** * A group that has never registered has no retry topic, so the lookup fails with @@ -193,4 +196,44 @@ class RocketMqAdminUtilTest { Assertions.assertEquals(Collections.singletonMap(messageQueue, 42L), offsets); } + + /** + * RocketMqSourceOptions.TOPICS accepts a comma separated list, so the lookup loops over several + * topics and merges their consume stats. Pinning that here keeps the loop from being rewritten + * into one that returns the first topic's offsets and drops the rest. + */ + @Test + void testCurrentOffsets_multipleTopicsAreMergedAcrossTheLoop() throws Exception { + MessageQueue firstQueue = new MessageQueue(TOPIC, "broker-a", 0); + MessageQueue secondQueue = new MessageQueue(OTHER_TOPIC, "broker-a", 0); + + DefaultMQAdminExt adminClient = Mockito.mock(DefaultMQAdminExt.class); + Mockito.when(adminClient.examineConsumeStats(GROUP, TOPIC)) + .thenReturn(consumeStatsFor(firstQueue, 42L)); + Mockito.when(adminClient.examineConsumeStats(GROUP, OTHER_TOPIC)) + .thenReturn(consumeStatsFor(secondQueue, 7L)); + + Map<MessageQueue, Long> offsets = + RocketMqAdminUtil.currentOffsets( + adminClient, + GROUP, + Arrays.asList(TOPIC, OTHER_TOPIC), + new HashSet<>(Arrays.asList(firstQueue, secondQueue))); + + Map<MessageQueue, Long> expected = new HashMap<>(); + expected.put(firstQueue, 42L); + expected.put(secondQueue, 7L); + Assertions.assertEquals(expected, offsets); + } + + private static ConsumeStats consumeStatsFor(MessageQueue messageQueue, long committedOffset) { + OffsetWrapper offsetWrapper = new OffsetWrapper(); + offsetWrapper.setConsumerOffset(committedOffset); + HashMap<MessageQueue, OffsetWrapper> offsetTable = new HashMap<>(); + offsetTable.put(messageQueue, offsetWrapper); + + ConsumeStats consumeStats = new ConsumeStats(); + consumeStats.setOffsetTable(offsetTable); + return consumeStats; + } }
