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-12349-ea3df1667719520c51d40546d04f50912ba24313 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit b70652be8af46c0e98a0a2fc714adae249972ea4 Author: Sepuri Sai Krishna <[email protected]> AuthorDate: Mon Sep 21 05:24:17 2026 +0000 [Fix][Connector-V2] Surface RocketMQ group-offset lookup failures instead of rewinding (#12349) --- .../rocketmq/common/RocketMqAdminUtil.java | 122 ++++++++++--- .../rocketmq/common/RocketMqAdminUtilTest.java | 196 +++++++++++++++++++++ .../source/RocketMqSourceSplitEnumeratorTest.java | 51 ++++++ 3 files changed, 341 insertions(+), 28 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 c00b568986..589d917cbc 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 @@ -47,6 +47,8 @@ import org.apache.rocketmq.remoting.protocol.LanguageCode; import org.apache.rocketmq.tools.admin.DefaultMQAdminExt; import org.apache.rocketmq.tools.command.CommandUtil; +import lombok.extern.slf4j.Slf4j; + import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -58,6 +60,7 @@ import java.util.UUID; import java.util.stream.Collectors; /** Tools for creating RocketMq topic and group. */ +@Slf4j public class RocketMqAdminUtil { public static String createUniqInstance(String prefix) { @@ -273,47 +276,110 @@ public class RocketMqAdminUtil { } } - /** Get consumer group offset */ + /** + * Get consumer group offset. + * + * <p>An empty result means the lookup succeeded and this group has committed nothing for the + * requested queues. A failed lookup is never reported that way: callers such as {@code + * RocketMqSourceSplitEnumerator}'s {@code CONSUME_FROM_GROUP_OFFSETS} branch read an empty map + * as "this group has committed nothing" and rewind to the first offset, so mapping a failure + * onto that answer would silently re-deliver a whole topic. + * + * <p>{@code TOPIC_NOT_EXIST} (code 17) needs care, because it is the only failure code this + * call can raise for a route problem and it covers two very different situations. See {@link + * #currentOffsets(DefaultMQAdminExt, String, List, Set)} for how they are told apart. + */ public static Map<MessageQueue, Long> currentOffsets( RocketMqBaseConfiguration config, List<String> topics, Set<MessageQueue> messageQueues) { - // Get consumer group offset DefaultMQAdminExt adminClient = null; try { adminClient = RocketMqAdminUtil.startMQAdminTool(config); - Map<MessageQueue, OffsetWrapper> consumerOffsets = Maps.newConcurrentMap(); - for (String topic : topics) { - ConsumeStats consumeStats = - adminClient.examineConsumeStats(config.getGroupId(), topic); - consumerOffsets.putAll(consumeStats.getOffsetTable()); + return currentOffsets(adminClient, config.getGroupId(), topics, messageQueues); + } catch (MQClientException e) { + throw new RocketMqConnectorException( + RocketMqConnectorErrorCode.GET_CONSUMER_GROUP_OFFSETS_ERROR, e); + } finally { + if (adminClient != null) { + adminClient.shutdown(); } - return consumerOffsets.keySet().stream() - .filter(messageQueue -> messageQueues.contains(messageQueue)) - .collect( - Collectors.toMap( - messageQueue -> messageQueue, - messageQueue -> - consumerOffsets.get(messageQueue).getConsumerOffset())); - } catch (MQClientException - | MQBrokerException - | RemotingException - | InterruptedException e) { - if (e instanceof MQClientException) { - if (((MQClientException) e).getResponseCode() == ResponseCode.TOPIC_NOT_EXIST) { + } + } + + /** + * Collects committed group offsets through an already-started admin client. Package-private so + * that tests can drive {@link DefaultMQAdminExt#examineConsumeStats} deterministically. + * + * <p>{@code examineConsumeStats} resolves its route solely from the group's auto-generated + * retry topic, never from the requested topic, so a {@code TOPIC_NOT_EXIST} response always + * refers to that retry topic. It is raised both when the group has never registered, in which + * case the retry topic has not been created yet, and when a route has been lost, in which case + * the offsets exist but are briefly unreadable. Answering "nothing committed" is correct for + * the first and causes a full replay for the second. + * + * <p>The two are separated by re-resolving the requested topic. The name server removes routes + * per broker rather than per topic, so an outage takes the requested topic's route along with + * 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. + */ + static Map<MessageQueue, Long> currentOffsets( + DefaultMQAdminExt adminClient, + String groupId, + List<String> topics, + Set<MessageQueue> messageQueues) { + Map<MessageQueue, OffsetWrapper> consumerOffsets = Maps.newConcurrentMap(); + for (String topic : topics) { + try { + ConsumeStats consumeStats = adminClient.examineConsumeStats(groupId, topic); + consumerOffsets.putAll(consumeStats.getOffsetTable()); + } 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. + 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.", + groupId, + topic); return Collections.emptyMap(); - } else { - throw new RocketMqConnectorException( - RocketMqConnectorErrorCode.GET_CONSUMER_GROUP_OFFSETS_ERROR, e); } - } else { + throw new RocketMqConnectorException( + RocketMqConnectorErrorCode.GET_CONSUMER_GROUP_OFFSETS_ERROR, e); + } catch (MQBrokerException | RemotingException e) { + throw new RocketMqConnectorException( + RocketMqConnectorErrorCode.GET_CONSUMER_GROUP_OFFSETS_ERROR, e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); throw new RocketMqConnectorException( RocketMqConnectorErrorCode.GET_CONSUMER_GROUP_OFFSETS_ERROR, e); } - } finally { - if (adminClient != null) { - adminClient.shutdown(); - } + } + return consumerOffsets.keySet().stream() + .filter(messageQueue -> messageQueues.contains(messageQueue)) + .collect( + Collectors.toMap( + messageQueue -> messageQueue, + messageQueue -> + consumerOffsets.get(messageQueue).getConsumerOffset())); + } + + /** + * Reports whether the name server can still resolve a route for {@code topic}. Any failure is + * reported as unavailable, since the caller only uses this to decide whether a route problem is + * broker wide. + */ + private static boolean topicRouteAvailable(DefaultMQAdminExt adminClient, String topic) { + try { + return adminClient.examineTopicRouteInfo(topic) != null; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } catch (MQClientException | RemotingException e) { + return false; } } } 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 new file mode 100644 index 0000000000..daae2bbcc7 --- /dev/null +++ b/seatunnel-connectors-v2/connector-rocketmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/common/RocketMqAdminUtilTest.java @@ -0,0 +1,196 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.rocketmq.common; + +import org.apache.seatunnel.connectors.seatunnel.rocketmq.exception.RocketMqConnectorException; + +import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.common.admin.ConsumeStats; +import org.apache.rocketmq.common.admin.OffsetWrapper; +import org.apache.rocketmq.common.message.MessageQueue; +import org.apache.rocketmq.common.protocol.ResponseCode; +import org.apache.rocketmq.common.protocol.route.TopicRouteData; +import org.apache.rocketmq.tools.admin.DefaultMQAdminExt; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +/** + * Covers the contract of {@link RocketMqAdminUtil#currentOffsets}: an empty result must mean the + * lookup succeeded and nothing was committed for the requested queues, never that the lookup + * failed. Callers rewind to the first offset on an empty map, so conflating the two silently + * re-delivers a whole topic. + * + * <p>{@code examineConsumeStats} resolves its route from the group's retry topic rather than from + * the requested topic, so {@code TOPIC_NOT_EXIST} covers both "this group has never registered" and + * "a route was lost". The two must be separated by re-resolving the requested topic, and the tests + * below pin both directions. + */ +class RocketMqAdminUtilTest { + + private static final String GROUP = "test-group"; + private static final String TOPIC = "test-topic"; + + /** + * A group that has never registered has no retry topic, so the lookup fails with + * TOPIC_NOT_EXIST while the requested topic still resolves. That is a genuine cold start and + * must return an empty map so the caller applies the configured start mode. + */ + @Test + void testCurrentOffsets_retryTopicMissingWhileRouteHealthyReturnsEmpty() throws Exception { + DefaultMQAdminExt adminClient = Mockito.mock(DefaultMQAdminExt.class); + 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()); + + MessageQueue messageQueue = new MessageQueue(TOPIC, "broker-a", 0); + + Map<MessageQueue, Long> offsets = + RocketMqAdminUtil.currentOffsets( + adminClient, + GROUP, + Collections.singletonList(TOPIC), + Collections.singleton(messageQueue)); + + Assertions.assertTrue(offsets.isEmpty()); + } + + /** + * The name server drops routes per broker, so a real outage takes the requested topic's route + * along with the retry topic's. Committed offsets may well exist, so this must surface as a + * failure rather than as "nothing committed". + */ + @Test + void testCurrentOffsets_routeOutageSurfacesAsFailure() throws Exception { + DefaultMQAdminExt adminClient = Mockito.mock(DefaultMQAdminExt.class); + 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)) + .thenThrow( + new MQClientException( + ResponseCode.TOPIC_NOT_EXIST, + "No topic route info in name server for the topic: " + TOPIC)); + + MessageQueue messageQueue = new MessageQueue(TOPIC, "broker-a", 0); + + Assertions.assertThrows( + RocketMqConnectorException.class, + () -> + RocketMqAdminUtil.currentOffsets( + adminClient, + GROUP, + Collections.singletonList(TOPIC), + Collections.singleton(messageQueue))); + } + + /** + * The route probe only applies to TOPIC_NOT_EXIST. Any other failure surfaces even when the + * requested topic resolves perfectly well. + */ + @Test + void testCurrentOffsets_otherResponseCodeSurfacesEvenWhenRouteHealthy() throws Exception { + DefaultMQAdminExt adminClient = Mockito.mock(DefaultMQAdminExt.class); + Mockito.when(adminClient.examineConsumeStats(GROUP, TOPIC)) + .thenThrow( + new MQClientException( + ResponseCode.CONSUMER_NOT_ONLINE, + "Not found the consumer group consume stats")); + Mockito.when(adminClient.examineTopicRouteInfo(TOPIC)).thenReturn(new TopicRouteData()); + + MessageQueue messageQueue = new MessageQueue(TOPIC, "broker-a", 0); + + Assertions.assertThrows( + RocketMqConnectorException.class, + () -> + RocketMqAdminUtil.currentOffsets( + adminClient, + GROUP, + Collections.singletonList(TOPIC), + Collections.singleton(messageQueue))); + } + + /** + * A successful lookup that matches none of the requested queues is the legitimate empty case, + * for example a newly discovered queue that has no entry in the consume stats yet. It must + * still return an empty map so the caller can cold-start that queue. + */ + @Test + void testCurrentOffsets_successfulLookupWithNoMatchingQueueReturnsEmpty() throws Exception { + MessageQueue requestedQueue = new MessageQueue(TOPIC, "broker-a", 1); + MessageQueue committedQueue = new MessageQueue(TOPIC, "broker-a", 0); + + OffsetWrapper offsetWrapper = new OffsetWrapper(); + offsetWrapper.setConsumerOffset(42L); + Map<MessageQueue, OffsetWrapper> offsetTable = new HashMap<>(); + offsetTable.put(committedQueue, offsetWrapper); + + ConsumeStats consumeStats = new ConsumeStats(); + consumeStats.setOffsetTable((HashMap<MessageQueue, OffsetWrapper>) offsetTable); + + DefaultMQAdminExt adminClient = Mockito.mock(DefaultMQAdminExt.class); + Mockito.when(adminClient.examineConsumeStats(GROUP, TOPIC)).thenReturn(consumeStats); + + Map<MessageQueue, Long> offsets = + RocketMqAdminUtil.currentOffsets( + adminClient, + GROUP, + Collections.singletonList(TOPIC), + Collections.singleton(requestedQueue)); + + Assertions.assertTrue(offsets.isEmpty()); + } + + /** A successful lookup that does match returns the committed offset unchanged. */ + @Test + void testCurrentOffsets_successfulLookupReturnsCommittedOffset() throws Exception { + MessageQueue messageQueue = new MessageQueue(TOPIC, "broker-a", 0); + + OffsetWrapper offsetWrapper = new OffsetWrapper(); + offsetWrapper.setConsumerOffset(42L); + Map<MessageQueue, OffsetWrapper> offsetTable = new HashMap<>(); + offsetTable.put(messageQueue, offsetWrapper); + + ConsumeStats consumeStats = new ConsumeStats(); + consumeStats.setOffsetTable((HashMap<MessageQueue, OffsetWrapper>) offsetTable); + + DefaultMQAdminExt adminClient = Mockito.mock(DefaultMQAdminExt.class); + Mockito.when(adminClient.examineConsumeStats(GROUP, TOPIC)).thenReturn(consumeStats); + + Map<MessageQueue, Long> offsets = + RocketMqAdminUtil.currentOffsets( + adminClient, + GROUP, + Collections.singletonList(TOPIC), + Collections.singleton(messageQueue)); + + Assertions.assertEquals(Collections.singletonMap(messageQueue, 42L), offsets); + } +} diff --git a/seatunnel-connectors-v2/connector-rocketmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/source/RocketMqSourceSplitEnumeratorTest.java b/seatunnel-connectors-v2/connector-rocketmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/source/RocketMqSourceSplitEnumeratorTest.java index 00bbb9ece9..2e2c185c75 100644 --- a/seatunnel-connectors-v2/connector-rocketmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/source/RocketMqSourceSplitEnumeratorTest.java +++ b/seatunnel-connectors-v2/connector-rocketmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/source/RocketMqSourceSplitEnumeratorTest.java @@ -20,6 +20,8 @@ package org.apache.seatunnel.connectors.seatunnel.rocketmq.source; import org.apache.seatunnel.api.source.SourceSplitEnumerator; import org.apache.seatunnel.connectors.seatunnel.rocketmq.common.RocketMqAdminUtil; import org.apache.seatunnel.connectors.seatunnel.rocketmq.common.StartMode; +import org.apache.seatunnel.connectors.seatunnel.rocketmq.exception.RocketMqConnectorErrorCode; +import org.apache.seatunnel.connectors.seatunnel.rocketmq.exception.RocketMqConnectorException; import org.apache.rocketmq.common.admin.TopicOffset; import org.apache.rocketmq.common.message.MessageQueue; @@ -44,6 +46,7 @@ import java.util.stream.Collectors; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.never; class RocketMqSourceSplitEnumeratorTest { @@ -313,6 +316,54 @@ class RocketMqSourceSplitEnumeratorTest { "New split must use the configured start mode offset"); } + /** + * A failed group-offset lookup must not be mistaken for "this group has committed nothing". + * Reporting it that way would send CONSUME_FROM_GROUP_OFFSETS down the first-offset fallback + * and silently re-deliver the whole topic, so the failure has to propagate instead. + */ + @Test + void testRun_doesNotFallBackToFirstOffsetWhenGroupOffsetLookupFails() { + ConsumerMetadata metadata = new ConsumerMetadata(); + metadata.setTopics(Collections.singletonList("topic_group")); + metadata.setStartMode(StartMode.CONSUME_FROM_GROUP_OFFSETS); + + MessageQueue messageQueue = new MessageQueue("topic_group", "broker-group", 0); + + SourceSplitEnumerator.Context<RocketMqSourceSplit> context = mockContext(); + + try (MockedStatic<RocketMqAdminUtil> mockedAdmin = + Mockito.mockStatic(RocketMqAdminUtil.class)) { + mockedAdmin + .when(() -> RocketMqAdminUtil.offsetTopics(any(), eq(metadata.getTopics()))) + .thenReturn( + Collections.singletonList( + topicOffsets(queueOffset(messageQueue, 5L, 18L)))); + mockedAdmin + .when( + () -> + RocketMqAdminUtil.currentOffsets( + any(), + eq(metadata.getTopics()), + eq(Collections.singleton(messageQueue)))) + .thenThrow( + new RocketMqConnectorException( + RocketMqConnectorErrorCode.GET_CONSUMER_GROUP_OFFSETS_ERROR, + "simulated route lookup failure")); + + RocketMqSourceSplitEnumerator enumerator = + new RocketMqSourceSplitEnumerator( + metadata, Collections.emptyMap(), context, -1L); + + Assertions.assertThrows(RocketMqConnectorException.class, enumerator::run); + + mockedAdmin.verify( + () -> RocketMqAdminUtil.flatOffsetTopics(any(), eq(metadata.getTopics())), + never()); + } + + Mockito.verify(context, never()).assignSplit(Mockito.anyInt(), Mockito.anyList()); + } + private SourceSplitEnumerator.Context<RocketMqSourceSplit> mockContext() { SourceSplitEnumerator.Context<RocketMqSourceSplit> context = Mockito.mock(SourceSplitEnumerator.Context.class);
