lizhimins opened a new issue, #151: URL: https://github.com/apache/rocketmq-flink/issues/151
### Before Creating the Report - [x] I searched the existing issues and did not find the same issue. ### Environment - Apache RocketMQ server 5.5.1 (open-source distribution: NameServers + Brokers, **no proxy**), clients connecting to the NameServer addresses directly - Flink 1.20.3, `flink-connector-rocketmq` built from main (`7170770`), remoting module - Kubernetes acceptance run: 2 brokers × 2 queues per topic, source parallelism 1–4 Two independent defects showed up in this setup. Both are silent: jobs stay `RUNNING`, checkpoints succeed, and nothing is logged at ERROR level by the connector. ### Defect 1 — starting offsets are ignored for a random subset of queues (assign/seek races with the client-side first pull) With `latest`, `timestamp` or explicit-offset startup modes, roughly half of the queues replay history from offset 0 on every fresh assignment; the affected subset changes between runs. Root cause (verified by decompiling rocketmq-client 5.5.1): `RebalancePullImpl.computePullFromWhereWithException` is hard-coded to `return 0`. `DefaultLitePullConsumer.assign()` wakes the client-internal `RebalanceService` thread, which creates `PullTask`s for the newly assigned queues and dispatches a first pull from offset 0 immediately. The connector's `seek(startingOffset)` in `handleSplitsChanges` races with that first pull on the client's own threads — no Flink-side lock scope can prevent this, and presetting the offset store does not help because `computePullFromWhere` never reads it. Whoever loses the race either replays `[0, startingOffset)` or has its `nextBeginOffset` overwritten. Why proxy deployments never showed it: behind a proxy the route is a single virtual brokerName, so all queues share one rebalance path; with direct NameServer routes (multiple real brokerNames) the race surfaces per broker/queue. Suggested fix direction: enforce the starting offset deterministically at the `fetch()` outlet — drop records with `queueOffset < max(startingOffset, 0)` using the split's recorded starting offset — instead of relying on `seek()` winning a cross-thread race. `seek()` stays as a traffic optimization. ### Defect 2 — a failed route fetch is treated as "the route is now empty" (mass split revocation + offset rewind during a NameServer outage) Stopping all NameServers for 60s restarts the Flink job and, after recovery, replays committed data (observed: sink topic grew 200 → 350, i.e. +150 duplicated records). Root cause: `RocketMQSourceEnumerator.requestServiceDiscovery` catches per-topic `fetchMessageQueues` failures and flattens them into an empty queue set. Diffed against `allocatedSet` in `getSourceChangeResult`, **every allocated queue lands in `decreaseSet`**, triggering a full rebalance that revokes all splits; when the NameServer is back the same queues return as `increase` and go through `initializeSourceSplits` again — a mass revocation plus an offset rewind for a route that never actually changed. Note that the RocketMQ client already treats the NameServer as a weak dependency at runtime (`MQClientInstance.topicRouteTable` keeps the last known route when a refresh fails, and data-plane pulls continue). The connector does not need its own route cache — it only needs to stop interpreting "fetch failed" as "route empty". Suggested fix direction: exclude failed topics from the discovery round (they must not contribute decreases); skip the round entirely when every topic failed; defensively keep the current assignment when a discovery result is empty while queues are allocated. Genuine queue deletions surface again on later rounds, so nothing is masked permanently. ### Related documentation gap While tracing a separate transactional-sink issue we confirmed that `EXACTLY_ONCE` cannot work when endpoints point at a proxy that keeps transaction state only in per-process memory. The remoting connector docs should state this explicitly and point users to `AT_LEAST_ONCE` + idempotent consumers for proxy-fronted endpoints, while noting that direct deployments (NameServer endpoints) do support `EXACTLY_ONCE`. ### Acceptance evidence Cluster-level reproductions and verifications for both defects (including an A/B run of the NameServer-outage scenario before/after the fix: job `RESTARTING` + 150 replayed records → job `RUNNING` throughout + 0 replayed) are documented in our internal acceptance report; I can attach the raw logs to the PR if helpful. -- 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]
