SEPURI-SAI-KRISHNA commented on issue #12332:
URL: https://github.com/apache/seatunnel/issues/12332#issuecomment-5690466293
Following up on my own report with one refinement, because I went to write
the patch and found that the fix cannot go where it would be most convenient.
The tempting place is the call site, guarding the fallback in
`RocketMqSourceSplitEnumerator`:
```java
case CONSUME_FROM_GROUP_OFFSETS:
Map<MessageQueue, Long> groupOffsets = listConsumerGroupOffsets(queues);
if (groupOffsets.isEmpty()) { ... first offset ... }
```
That does not work, because an empty map there has three different causes
and the enumerator cannot tell them apart:
1. the topic genuinely does not exist, so a cold start is correct;
2. the admin lookup failed transiently on a missing route, which is the
defect in this issue;
3. the lookup **succeeded** and the filter at the end of `currentOffsets`,
`.filter(messageQueue -> messageQueues.contains(messageQueue))`, removed
everything, which happens for a newly discovered queue that has no entry in the
consume stats yet.
Case 3 is legitimate and common with partition discovery enabled, and
starting such a queue from the first offset is the right behavior. Any guard at
the call site that refuses to fall back on an empty map would break it. So the
distinction has to be drawn inside `currentOffsets`, which is the only place
that knows whether the admin call threw or returned and was filtered.
That constrains the fix in a way worth agreeing before I write it. Inside
`currentOffsets` the two exception cases are separable, since
`RocketMqAdminUtil.topicExist` at line 186 answers exactly the question that
splits case 1 from case 2, and the existing behavior is preserved either way: a
genuinely absent topic still returns an empty map and still cold-starts, while
an existing topic whose route is briefly unreadable stops being reported as
"nothing committed". Case 3 is untouched because it never enters the catch
block.
The open question is what should happen in case 2 once it is
distinguishable: retry the lookup and then fail if it does not recover, or fail
immediately. Failing is safe in the sense that rewinding and silently
re-delivering a topic is worse than a visible error, but it does turn a
transient condition into a job failure, so it is a judgement call rather than a
purely mechanical fix.
There is also a testability constraint I would rather surface than work
around quietly. `currentOffsets` builds its admin client internally through
`startMQAdminTool(config)`, so there is no seam to inject a failing client in a
unit test. `RocketMqSourceSplitEnumeratorTest` already uses
`Mockito.mockStatic(RocketMqAdminUtil.class)`, but that mocks the very class
the fix would live in, so it cannot cover this. Introducing a small seam would
make the fix properly testable, at the cost of touching a little more than the
defect strictly requires.
I am happy to implement whichever combination is preferred. I am holding off
on the PR only because the retry-versus-fail choice and the seam are decisions
I would rather not make unilaterally in production connector code on a
data-correctness path.
--
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]
