SEPURI-SAI-KRISHNA commented on issue #12332:
URL: https://github.com/apache/seatunnel/issues/12332#issuecomment-5697682577
Thank you, that is a better contract than the one I proposed, and your
correction kills my suggested approach outright.
I checked `topicExist` before accepting it and you are right:
`RocketMqAdminUtil.java:196-198` catches `MQClientException`, tests for
`ResponseCode.TOPIC_NOT_EXIST` and returns `false`. It carries exactly the same
ambiguity as `currentOffsets`, so it cannot prove a topic is genuinely absent
and cannot be used to separate the cases. The "Possible fix" section of the
description above is therefore wrong, and I would rather flag that here than
leave it for someone to build from.
One piece of supporting evidence for your contract that I had not noticed
when I filed this. In the enumerator path the topic is already proven to exist
before `currentOffsets` is ever called:
```java
public void run() throws Exception {
...
fetchPendingPartitionSplit(); // -> getTopicInfo() ->
RocketMqAdminUtil.offsetTopics(...)
setPartitionStartOffset(); // -> listConsumerGroupOffsets() ->
currentOffsets(...)
```
`RocketMqSourceSplitEnumerator.java:177-180`. `offsetTopics` has no
`TOPIC_NOT_EXIST` branch at all and raises `GET_MIN_AND_MAX_OFFSETS_ERROR` for
a topic it cannot resolve, so a genuinely absent topic fails there first and
never reaches the group-offset lookup. That means the `TOPIC_NOT_EXIST` branch
in `currentOffsets` is only ever reachable from a transient failure on this
path, which is the argument that the change cannot break a real cold start
rather than merely trading one risk for another.
On the E2E regression, your condition was "if the route-error scenario can
be injected without timing dependence". I do not think it can, and I would
rather say so than add a test that reintroduces the class of problem I have
been fixing in #12323.
The injection itself is available. `RocketMqIT.deleteTopicIfExist` already
does two separable things:
```java
admin.deleteTopicInBroker(brokerNames, topicName);
admin.deleteTopicInNameServer(brokerNames, topicName, "delete_topic");
```
Calling only the second removes the route while leaving the broker's topic
data and committed group offsets intact, which is exactly this bug's
precondition. The problem is the window: brokers re-register their topics with
the name server on a periodic heartbeat, so the route returns on its own and
the test would be racing that interval. A regression test whose correctness
depends on beating a heartbeat is precisely the shape of flakiness I would be
adding back. So I am leaving the E2E out and relying on the deterministic unit
coverage, unless you would rather have it with that caveat stated in the test.
What I am building, to confirm I have understood the contract:
- `currentOffsets` returns an empty map only when `examineConsumeStats`
completed normally and the filter matched nothing. Every caught exception,
`TOPIC_NOT_EXIST` included, becomes `RocketMqConnectorException`.
- `InterruptedException` re-interrupts the current thread before being
wrapped.
- No retry policy, no new option.
- The smallest seam I could find is a package-private overload taking an
already-started `DefaultMQAdminExt`, so a test can drive `examineConsumeStats`
with Mockito. No mutable static state and no change to the public signature.
- Enumerator coverage for your three points. Two of them are already covered
by existing tests in `RocketMqSourceSplitEnumeratorTest` that must keep
passing, `testRun_fallsBackToFirstOffsetWhenGroupOffsetsAreMissing` for the
successful filtered empty case and `testRestore_preservesCheckpointOffsets`
plus `testRestore_newQueueUsesConfiguredStartMode` for the restore path, so the
new one is the failure case: a code 17 lookup must propagate and must not reach
the first-offset fallback.
One judgement call I would like checked. `RocketMqAdminUtil` has no test
class today, so covering the new contract at the unit level means adding one,
which sits against the "do not create new test classes casually" guidance. I
did not think putting `RocketMqAdminUtil` assertions inside the enumerator's
test class was better. Happy to move them if you disagree.
--
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]
