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 36c0d1b8a6 [Fix][E2E] Retry and route-guard the remaining RocketMqIT
offset reads (#12392)
36c0d1b8a6 is described below
commit 36c0d1b8a6fc880f208384268f7bec435e7bf601
Author: Sepuri Sai Krishna <[email protected]>
AuthorDate: Mon Sep 21 05:26:22 2026 +0000
[Fix][E2E] Retry and route-guard the remaining RocketMqIT offset reads
(#12392)
---
.../e2e/connector/rocketmq/RocketMqIT.java | 35 ++++++++++++++++++----
1 file changed, 30 insertions(+), 5 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java
index 8bbcd2d986..6017b05f90 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java
@@ -416,6 +416,7 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
String topicName = "test_topic_message_tag";
String tag = "test_tag";
+ waitForTopicRoute(topicName);
Map<String, RocketMqConsumerMessage> data =
getRocketMqConsumerData(topicName);
ObjectMapper objectMapper = new ObjectMapper();
String key = data.keySet().iterator().next();
@@ -531,12 +532,26 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
exception -> exception instanceof
RocketMqConnectorException,
Constant.OPERATION_RETRY_SLEEP));
consumer.assign(queueOffsets.keySet());
- // seek to offset
+ // seek to offset. Retry-wrapped like the offsetTopics lookup
above, since both read
+ // broker metadata that can be briefly unavailable. On dev
currentOffsets still maps a
+ // failed lookup onto an empty map, so today this is a consistency
fix; it becomes load
+ // bearing once #12349 makes that call raise
RocketMqConnectorException, which is the
+ // exception this predicate matches.
+ // shouldThrowException is true here, unlike the lookup above,
because RetryUtils
+ // returns null once retries are exhausted when it is false, and a
null map would
+ // surface as a bare NPE on the next line with the real cause
discarded.
Map<MessageQueue, Long> currentOffsets =
- RocketMqAdminUtil.currentOffsets(
- newConfiguration(),
- Lists.newArrayList(topicName),
- queueOffsets.keySet());
+ RetryUtils.retryWithException(
+ () ->
+ RocketMqAdminUtil.currentOffsets(
+ newConfiguration(),
+ Lists.newArrayList(topicName),
+ queueOffsets.keySet()),
+ new RetryUtils.RetryMaterial(
+ Constant.OPERATION_RETRY_TIME,
+ true,
+ exception -> exception instanceof
RocketMqConnectorException,
+ Constant.OPERATION_RETRY_SLEEP));
for (MessageQueue mq : queueOffsets.keySet()) {
long currentOffset =
currentOffsets.containsKey(mq)
@@ -621,6 +636,16 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
private void checkOffsetNoDiff(String topicName, String consumerGroup) {
RocketMqBaseConfiguration config = newConfiguration();
config.setGroupId(consumerGroup);
+ // Confirms the route for the offsetTopics read below. currentOffsets
resolves
+ // %RETRY%<group> rather than this topic, which the call does not
create or check, but the
+ // name server drops routes per broker rather than per topic, so a
resolvable route here
+ // means the broker registration is live and the retry topic's route
with it. That topic
+ // exists by this point because the job has already consumed with this
group.
+ // This covers a brief gap, not a long one: waitForTopicRoute gives up
after a minute, so
+ // the 4m24s outage measured in CI would still fail, just with a route
message instead of
+ // an opaque assertion timeout. The 30 second assertion window is
deliberately left alone
+ // so a genuine offset mismatch keeps failing fast.
+ waitForTopicRoute(topicName);
Awaitility.await()
.ignoreExceptions()
.atMost(30, TimeUnit.SECONDS)
