Copilot commented on code in PR #12340:
URL: https://github.com/apache/seatunnel/pull/12340#discussion_r4022998314
##########
seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/common/RocketMqAdminUtil.java:
##########
@@ -284,9 +283,24 @@ public static Map<MessageQueue, Long> currentOffsets(
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());
+ try {
+ ConsumeStats consumeStats =
+
adminClient.examineConsumeStats(config.getGroupId(), topic);
+ consumerOffsets.putAll(consumeStats.getOffsetTable());
+ } catch (MQClientException e) {
+ if (e.getResponseCode() == ResponseCode.TOPIC_NOT_EXIST) {
+ if (topicExist(config, topic)) {
+ throw new RocketMqConnectorException(
+
RocketMqConnectorErrorCode.GET_CONSUMER_GROUP_OFFSETS_ERROR, e);
+ }
Review Comment:
`topicExist` cannot distinguish the transient case this change is intended
to handle: it calls `examineTopicRouteInfo` and maps the same `TOPIC_NOT_EXIST`
response code to `false` at `RocketMqAdminUtil.java:190-198`. If the main
topic's route is also temporarily unavailable, this branch still falls through
and returns an empty map, so the caller can still interpret the lookup failure
as "no committed offsets" and start from the first offset. Please avoid using
this boolean re-check as proof of nonexistence; retry or propagate an unknown
result instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
