lizhimins opened a new pull request, #152:
URL: https://github.com/apache/rocketmq-flink/pull/152
## What is the purpose of the change
Fixes the two silent defects reported in #151, both observed during an
acceptance run against an open-source RocketMQ 5.5.1 cluster with clients
connecting to NameServers directly (no proxy):
1. **Starting-offset race (D24)** —
`RebalancePullImpl.computePullFromWhereWithException` in rocketmq-client 5.5.1
is hard-coded to `return 0`, so `assign()` lets the client-internal rebalance
thread dispatch a first pull from offset 0 while `handleSplitsChanges` is still
seeking to the configured starting offset. With `latest` / `timestamp` /
explicit-offset modes, a random subset of queues (~half per run, subset varies)
replays history. No Flink-side lock scope can win this race — the pull happens
on client-internal threads, and presetting the offset store does not help
because `computePullFromWhere` never reads it (both alternatives were tried and
verified ineffective).
**Fix:** enforce the starting offset deterministically at the `fetch()`
outlet — records with `queueOffset < max(startingOffset, 0)` are dropped using
the split's recorded starting offset. `seek()` remains as a traffic-reduction
optimization.
2. **Route-fetch failure misread as an empty route (D26)** — during a
NameServer outage every `fetchMessageQueues` call fails and the periodic
discovery flattens the failures into an empty queue set; diffed against the
allocated set, that revokes **every** split, and the queues returning after
recovery go through `initializeSourceSplits` again (observed: job restart + 150
replayed records). The client already treats the NameServer as a weak
dependency (`topicRouteTable` keeps the last route on refresh failure), so the
connector needs no route cache of its own — only the failure semantics fixed
here.
**Fix:** failed topics are excluded from the round (they cannot
contribute decreases); a round where every topic failed is skipped; an empty
result while queues are allocated keeps the current assignment. Genuine
deletions surface again on later rounds.
Also documents that `EXACTLY_ONCE` is not supported on proxy-fronted
endpoints (transaction state lives in per-proxy-process memory and commits can
be silently dropped), while direct deployments support it.
## Brief change log
- `RocketMQSplitReader.fetch()`: drop records below the split's starting
offset at the outlet.
- `RocketMQSourceEnumerator`: per-topic failure isolation in
`requestServiceDiscovery`, null round on total failure, empty-set guard in
`handleSourceQueueChange`.
- New tests: `RocketMQSourceEnumeratorDiscoveryToleranceTest` (5 cases,
including documenting the empty-set diff hazard); existing split-reader tests
cover the offset filter.
- `docs/remoting-connector.md`: EXACTLY_ONCE support matrix per endpoint
type.
## Verifications
- Module build green: 143 tests, 0 failures (spotless applied).
- Cluster A/B for the outage scenario (2 brokers × 2 queues, ns ×1 stopped
60s):
- before: job `RESTARTING`, sink 200 → 350 (+150 replayed);
- after: job `RUNNING` in 6/6 samples during the outage, guard log every
discovery round, zero decreases, sink grew exactly by the second wave (202 =
101 + 101, **0 replayed**), producer recovered within 150s of the NameServer
returning.
- Cluster three-tier re-verification for the starting-offset fix: `latest` /
`timestamp` / `specific=10` all show 0 out-of-range records and sink physical
offsets exactly 101 per tier.
Closes #151
--
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]
