lizhimins opened a new issue, #153: URL: https://github.com/apache/rocketmq-flink/issues/153
### Before Creating the Report - [x] I searched the existing issues and did not find the same issue. ### Environment - Apache RocketMQ 5.5.1 (open-source NameServers + Brokers, direct connect) and a commercial 5.5.1-based distribution behind proxies, both on Kubernetes - Flink 1.20.3, Application mode, `flink-connector-rocketmq` built from main (`80a98ac`), remoting module - Acceptance matrix: 4 KiB messages, TaskManager 2 cores / 2 slots, TM 1 / 2 / 4 / 8, 32 queues per topic ### Defect 1 — the "async" sink send runs synchronously on the task thread, capping throughput at one RTT per record `InnerProducerImpl.send()` wraps the blocking `producer.send(message)` in `CompletableFuture.supplyAsync(..., Runnable::run)`. `Runnable::run` executes the supplier on the calling thread, so every record is sent synchronously on the Flink task thread and the returned future is already completed. Consequences: - Single-slot throughput is bounded by `1 / send RTT`, not by CPU or by the broker. Observed: Proxy-architecture Sink TM8 = 16,397 rec/s with TM CPU at only 7.9% and zero CFS throttling, send RTT avg 0.97 ms (16 slots × ~1,030/s ≈ 16.5k). Adding TM CPU does not help. - `send.pending.max` is declared and documented as bounding in-flight async sends, but the semaphore can never accumulate permits because there is nothing in flight — the option is dead. - `RocketMQWriter.flush()` waits on `pendingRecords`, which is decremented in `whenComplete` of an already-completed future, so flush never actually waits for broker acknowledgements; `close()` does not flush at all. - A send that throws synchronously leaks the acquired permit, and `checkAsyncException()` uses `getAndSet(null)`, so an async failure is consumed once and then silently lost while the writer keeps accepting records. Suggested fix direction: use the real async `producer.send(message, SendCallback)` path on a bounded sender executor, complete the future only from the callback (failing it on non-`SEND_OK` results and synchronous submission errors), return permits on every failure path, keep async failures sticky, bound flush by `producer.send.timeout`, and drain pending sends on close. ### Defect 2 — startup seek fails the job on transient transport errors `InnerConsumerImpl.seek()` wraps every `MQClientException` into `FlinkRuntimeException`. During job startup against a freshly restarted or briefly unreachable cluster, `seek` (and the `minOffset` initialization behind it) can fail with `RemotingSendRequestException` / `RemotingConnectException` / `RemotingTimeoutException`. Observed: the exception exhausted the 3-restart budget and the job reached a terminal `FAILED` state, although the same cluster served reads and writes normally seconds later. Suggested fix direction: retry `seek` a bounded number of times (e.g. 2 retries, 100 ms backoff) only when the direct cause is one of the three transport exceptions above; propagate authentication, illegal-offset and other steady-state failures immediately, and preserve interruption. ### Acceptance evidence Both defects were reproduced and verified fixed in a Kubernetes acceptance run (4 KiB, TM 1 / 2 / 4 / 8, 180 s warmup / 300 s sampling per scenario): - Sink throughput before → after (Proxy architecture, TM8): 16,397 rec/s at 7.9% TM CPU → 200,301 rec/s at ~100% TM CPU; direct-connect Sink TM8 reaches 193,513 rec/s. Send completion RT avg ~0.8–2 ms, failure callbacks 0. - Source TM8 startup: before, `seek` transport failures exhausted the restart budget (terminal FAILED); after, complete 302 s windows with 0 job restarts and 0 container restarts. Raw logs and per-scenario JSON can be attached 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]
