This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 6934fd0062 [Improve][Connector-V2] Note the multi-topic early return
in currentOffsets (#12417)
6934fd0062 is described below
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;
+ }
}
