leizhiyuan opened a new pull request, #1457: URL: https://github.com/apache/rocketmq-clients/pull/1457
### Which Issue(s) This PR Fixes Fixes #1446 ### Brief Description The standard (non-FIFO) consumer worker waited for every ack / nack round trip before consuming the next message, so a worker was capped at `1 / (consume time + ack RTT)`. Java acks asynchronously (`ProcessQueueImpl#eraseMessage`). - `AckEntryProcessor::ack_message_in_background` / `change_invisible_duration_in_background` send the request on a copy of the processor in a background task. Failed requests still go to the retry queue of the worker. - At most `MAX_IN_FLIGHT_ACKS` (64) requests are in flight per worker; above that the worker waits, so a slow server slows consumption down instead of piling up requests. - `AckEntryProcessor::shutdown` waits for the in-flight requests before it flushes and stops the retry queue, so no ack is lost on shutdown. - Without the ack wait, the worker no longer suspends inside a batch. It yields once it held the runtime thread for 1 ms, so a slow listener does not hold a runtime thread for a whole batch. (Yielding after every message cost ~30% throughput in the benchmark because each yield polls the I/O driver.) - FIFO is unchanged on purpose: like Java `FifoConsumeService`, the next message of a group is consumed only after the ack of the previous one completed. #### Notes for reviewers - The worker tests that expected acks / nacks on the processor's own mock session now record the requests sent on any shadow session (`recording_session`) and assert after `ack_processor.shutdown()`: a failed mock expectation inside a background task would only panic that task. - Conflicts with #1447's PR (the consume call in the worker) and #1449's PR (`AckEntryProcessor`). I will rebase this one last; see #1445. ### How Did You Test This Change? **Unit tests**, both fail on `master`: - `test_standard_consume_does_not_wait_for_ack_round_trip`: 10 messages, ack RTT 50 ms; the worker must finish in < 150 ms (master: 501 ms), and all 10 acks must be sent by `shutdown`. - `test_standard_consume_yields_between_messages`: 1 runtime thread, 64 messages, listener blocks 5 ms; another task must run within 40 ms (without the yield: 326 ms). **End-to-end** (mock proxy over real gRPC, 4 queues × 4 workers, batch 32, ack and receive RTT 2 ms, listener ~0, 3 runs): **3,395 → 28,717 msg/s**; acked msg/s equals consumed msg/s within 0.1%. ``` $ cargo test # lib: 152 passed, 0 failed (master: 150) $ cargo clippy --all-features -- -D warnings # clean $ cargo fmt --check # clean $ RUSTDOCFLAGS='--cfg docsrs' cargo doc --no-deps --all-features # ok $ cargo build --examples --all-features # ok $ cargo +1.74.0 check --all-features --tests --examples # ok (MSRV) ``` --- Part of #1445 (overview, end-to-end benchmark and merge order). -- 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]
