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-12648-ee90b50012d44d7b96ee47f07afe995a475a79d2 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 07f32fffbda5452bbf4a3f4dc6de5e36ca69e5ab Author: Sepuri Sai Krishna <[email protected]> AuthorDate: Thu Oct 8 13:35:46 2026 +0000 [Fix][Connector-V2] Keep RocketMQ group offsets when a later topic has no retry route (#12648) --- .../rocketmq/common/RocketMqAdminUtil.java | 47 +++++++------- .../rocketmq/common/RocketMqAdminUtilTest.java | 74 ++++++++++++++++++++++ 2 files changed, 97 insertions(+), 24 deletions(-) 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 03ecf07937..f64bfdd0d3 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 @@ -50,7 +50,6 @@ import org.apache.rocketmq.tools.command.CommandUtil; import lombok.extern.slf4j.Slf4j; import java.util.Collection; -import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -323,6 +322,9 @@ public class RocketMqAdminUtil { * the retry topic's, while a group that has simply never registered leaves it intact. A group * also cannot commit an offset without first registering, which is what creates the retry * topic, so a missing retry topic on a healthy name server implies nothing was committed. + * + * <p>A topic answering that way is skipped rather than ending the lookup, so offsets already + * collected for earlier topics in a multi-topic request survive. */ static Map<MessageQueue, Long> currentOffsets( DefaultMQAdminExt adminClient, @@ -337,33 +339,30 @@ public class RocketMqAdminUtil { } catch (MQClientException e) { if (e.getResponseCode() == ResponseCode.TOPIC_NOT_EXIST && 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. + // The group's retry topic is missing while the requested topic still + // resolves, so nothing has been committed and this topic contributes no + // offsets. + // + // Skip it rather than returning, so that offsets already collected for + // earlier topics in the list survive. topics is a supported multi-topic list + // (RocketMqSourceOptions.TOPICS), and the guard probes the requested topic + // rather than the group's retry topic, so on a multi-broker cluster where the + // retry topic and a later topic are hosted on different brokers, losing the + // retry topic's broker part way through the loop still leaves the later topic + // resolvable. Returning there would discard the earlier offsets, and the + // caller reads an empty map as a cold start and rewinds every topic to its + // first offset. // - // 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 still empty here. The invariant does not cover one transition, and - // that transition is reachable today. The guard above probes the - // requested topic, not the group's retry topic, and every iteration - // re-resolves the retry topic. On a multi-broker cluster where the retry - // topic and a later topic in the list are hosted on different brokers, - // losing the retry topic's broker mid-loop leaves the later topic - // resolvable, so the guard passes and this return discards the offsets - // already collected for the earlier topics. The caller reads an empty map - // as a cold start and rewinds every topic to its first offset, so - // supporting that layout means turning this into a continue that keeps - // the earlier offsets. + // When every topic answers this way the map stays empty, which is the + // cold-start answer the caller needs. log.warn( - "Consumer group {} has no retry topic yet, so it has never registered " - + "and has committed nothing. Topic {} still resolves, so this " - + "is not a route outage.", + "Consumer group {} has no retry topic yet, so it has committed nothing " + + "for topic {}. That topic still resolves, so this is not a " + + "route outage; skipping it and keeping any offsets already " + + "collected.", groupId, topic); - return Collections.emptyMap(); + continue; } throw new RocketMqConnectorException( RocketMqConnectorErrorCode.GET_CONSUMER_GROUP_OFFSETS_ERROR, e); 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 6072ddc389..9ec0458fe2 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 @@ -226,6 +226,80 @@ class RocketMqAdminUtilTest { Assertions.assertEquals(expected, offsets); } + /** + * The guard probes the requested topic, not the group's retry topic. On a multi-broker cluster + * the retry topic and a later topic in the list can live on different brokers, so losing the + * retry topic's broker part way through the loop leaves the later topic resolvable: the guard + * passes even though offsets for the earlier topics have already been collected. Discarding + * them makes the caller read a cold start and rewind every topic to its first offset, so the + * offsets gathered before the failure must survive. + */ + @Test + void testCurrentOffsets_laterTopicFailureKeepsOffsetsFromEarlierTopics() throws Exception { + MessageQueue firstQueue = new MessageQueue(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)) + .thenThrow( + new MQClientException( + ResponseCode.TOPIC_NOT_EXIST, + "No topic route info in name server for the topic: %RETRY%" + + GROUP)); + Mockito.when(adminClient.examineTopicRouteInfo(OTHER_TOPIC)) + .thenReturn(new TopicRouteData()); + + Map<MessageQueue, Long> offsets = + RocketMqAdminUtil.currentOffsets( + adminClient, + GROUP, + Arrays.asList(TOPIC, OTHER_TOPIC), + new HashSet<>(Arrays.asList(firstQueue))); + + Assertions.assertEquals( + Collections.singletonMap(firstQueue, 42L), + offsets, + "the offset already collected for the first topic must survive a later topic's " + + "missing retry route"); + } + + /** + * The cold start contract, pinned for a multi-topic list rather than a single topic. Skipping + * an unresolved topic instead of returning keeps the loop running to the end, so "every topic + * has nothing committed" is now the sum of several skips rather than one early return. The + * caller reads an empty map as a cold start and applies the configured start mode, so the + * result must still be exactly empty when no topic contributes an offset. + */ + @Test + void testCurrentOffsets_everyTopicMissingRetryRouteStillReportsAColdStart() 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); + for (String topic : Arrays.asList(TOPIC, OTHER_TOPIC)) { + Mockito.when(adminClient.examineConsumeStats(GROUP, topic)) + .thenThrow( + new MQClientException( + ResponseCode.TOPIC_NOT_EXIST, + "No topic route info in name server for the topic: %RETRY%" + + GROUP)); + Mockito.when(adminClient.examineTopicRouteInfo(topic)).thenReturn(new TopicRouteData()); + } + + Map<MessageQueue, Long> offsets = + RocketMqAdminUtil.currentOffsets( + adminClient, + GROUP, + Arrays.asList(TOPIC, OTHER_TOPIC), + new HashSet<>(Arrays.asList(firstQueue, secondQueue))); + + Assertions.assertTrue( + offsets.isEmpty(), + "a list in which every topic has no retry route must stay a cold start, not a " + + "partial resume"); + } + private static ConsumeStats consumeStatsFor(MessageQueue messageQueue, long committedOffset) { OffsetWrapper offsetWrapper = new OffsetWrapper(); offsetWrapper.setConsumerOffset(committedOffset);
