lizhimins opened a new pull request, #148: URL: https://github.com/apache/rocketmq-flink/pull/148
### Motivation Two silent-wrong-data defects in the legacy connector, both found by an acceptance run against a commercial 5.5.1 cluster (Flink 1.20.3, Application mode) and both in the family ISSUE #141 tracks: the job stays RUNNING, checkpoints succeed, nothing is logged, and the data is wrong. **1. The legacy source re-emits its whole topic, forever.** `pollAndEmit` read the next position back from `DefaultLitePullConsumer#committed` after every poll. Auto-commit is off by default, so the broker-side commit never advances and every round re-seeked to the same offset. Measured: 101 source messages became **120065 sink messages (118560 duplicates)** with the job RUNNING, zero restarts and zero errors. Two related faults fell out of fixing it: - `poll` drains **all** assigned queues, so seeking from one queue thread after another thread's poll rewound that queue and re-emitted what it had just drained. The first version of this fix still duplicated 30 messages for exactly this reason; seek + poll + booking are now atomic under the checkpoint lock. - `initOffsets` stored `committed()` verbatim, which is **-1** for a group that has never committed. Only the `GROUP_OFFSETS` branch had a fallback for that, so with `EARLIEST`/`LATEST` the first seek failed with `Seek offset illegal` and the source gave up, finishing the job having read nothing. It now falls back to the position the preceding seek left in the local offset store. **2. The legacy sink checkpoints over lost records.** An async send failure is parked and rethrown from `finish()`, and `invoke()` checks it too — but a burst of failures followed by an idle stream never reaches another `invoke()`, and `snapshotState` did not check at all. Measured: 101 rejected sends, 101 `Async send message failure!` errors, **0 task failures, 12 successful checkpoints, a clean stop-with-savepoint, and 0 records in the sink topic**. `snapshotState` now surfaces the parked failure, so a checkpoint cannot commit a loss. ### Modification - `RocketMQSourceFunction`: book the read position from the messages a poll drains (per queue, forward-only); make seek + poll + booking atomic; never store a negative start position. - `RocketMQSink`: call `checkAsyncSendException()` from `snapshotState()`. - Tests: `RocketMQSourceFunctionTest` gains a poll-position test (a stale `committed()` of 0 must not rewind a position of 6); new `RocketMQSinkAsyncFailureTest` pins that `snapshotState` fails the checkpoint when an async send already failed. ### Verification - `mvn -pl flink-connector-rocketmq test`: 123 tests, 0 failures, 0 errors. - Cluster acceptance, legacy source + legacy sink, 100 messages: duplicates **118560 → 0**, missing 0, unexpected 0, tail marker received with matching count. - Cluster acceptance, legacy sink in pure async mode with every send rejected (`CODE 13`): task failures **0 → 3**, stack top `IOException: Async send message failed` at `RocketMQSink.checkAsyncSendException` ← `snapshotState`, i.e. the loss is now loud instead of silent. -- 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]
