sun-9545sunoj opened a new pull request, #12340:
URL: https://github.com/apache/seatunnel/pull/12340
### Purpose of this pull request
This PR fixes a critical bug where RocketMqAdminUtil.currentOffsets treats
a transient metadata
lookup failure as proof that a consumer group has committed nothing,
causing the connector to
silently rewind and re-deliver the entire topic when start.mode =
CONSUME_FROM_GROUP_OFFSETS.
The Issue:
When adminClient.examineConsumeStats attempts to look up consumer group
offsets and the name
server temporarily has no route for the group (e.g., the %RETRY%GroupName
topic), it throws an
MQClientException with ResponseCode.TOPIC_NOT_EXIST. The connector was
unconditionally catching
this and returning an empty map. Because the core topic itself genuinely
exists, the subsequent
listOffsets call succeeds, setting the start offset to 0 and silently
rewinding the consumer.
The Fix:
Inside RocketMqAdminUtil.currentOffsets, we now catch the
MQClientException for each topic. If a
TOPIC_NOT_EXIST error is encountered, we explicitly check if the main
topic actually exists using
topicExist(config, topic).
• If topicExist is true, the failure is a transient consumer group
metadata lookup failure, so we
throw the RocketMqConnectorException to fail fast (and allow the engine to
retry).
• If topicExist is false, the topic genuinely does not exist, so we safely
return an empty map as
before.
(Resolves #12332)
### Does this PR introduce any user-facing change?
No. This restores the intended behavior of CONSUME_FROM_GROUP_OFFSETS by
preventing silent data
re-delivery during transient name server route unavailability.
### How was this patch tested?
The defect was established by logic and by CI logs demonstrating that
transient route
unavailability drops name server responses for minutes at a time. The fix
was tested locally to
ensure that if a topic physically exists (topicExist == true), the
connector will correctly throw
a GET_CONSUMER_GROUP_OFFSETS_ERROR instead of returning
Collections.emptyMap().
### Check list
[✓] If any new Jar binary package adding in your PR, please add License
Notice according New
License Guide
https://github.com/apache/seatunnel/blob/dev/docs/en/developer/new-license.md
(N/A -
no new dependencies)
[✓] If necessary, please update the documentation to describe the new
feature. (N/A - bug fix)
[✓] If necessary, please update incompatible-changes.md to describe the
incompatibility caused by
this PR. (N/A - behavior restored, no config changes)
[✓] If you are contributing the connector code, please check that the
following files are
updated:
1. Update plugin-mapping.properties and add new connector information
in it (N/A - existing
connector)
2. Update the pom file of seatunnel-dist (N/A)
3. Add ci label in label-scope-conf (N/A)
4. Add e2e testcase in seatunnel-e2e (N/A - transient name server
lookup failures are
difficult to simulate in standard E2E containers)
5. Update connector plugin_config (N/A)
--
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]
