fuyou001 commented on PR #11090: URL: https://github.com/apache/rocketmq/pull/11090#issuecomment-5611886271
Reviewed commit `b3c92c9d7058cb953c7917c28757ec181e169ce6`. I found two P1 issues in the newly supported external-executor path. Both are reproducible with a bounded `ThreadPoolExecutor` using the documented `AbortPolicy`. The Proxy's current unbounded executor does not encounter capacity rejection, but the public `setConsumeExecutor` API exposes these paths to ordinary consumers. ### [P1] Recover orderly consumption after executor rejection The [constructor now accepts the injected executor](https://github.com/apache/rocketmq/blob/b3c92c9d7058cb953c7917c28757ec181e169ce6/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java#L59-L62), but [submitConsumeRequest](https://github.com/apache/rocketmq/blob/b3c92c9d7058cb953c7917c28757ec181e169ce6/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java#L162-L173) has no rejection recovery. In the pull-success callback, `ProcessQueue.putMessage` sets `consuming=true` before submission. If the shared pool is full, submission throws. `MQClientAPIImpl` catches this and invokes `PullCallback.onException`, which schedules another pull; however, subsequent `putMessage` calls return `dispatchToConsume=false`, so no replacement consumption task is submitted. The queue remains stuck after executor capacity recovers and can eventually hit pull flow control. Scheduled continuations for lock retries, suspended consumption, and time-slice expiration reach the same submission method. Please retain and reschedule the rejected orderly request, including scheduled continuations, so a transient rejection cannot leave `consuming=true` without a consumption task. A regression test should saturate an injected pool, reject the first request, restore capacity, and deliver another batch; assert that consumption resumes in order, cached messages are removed, and offsets advance. Observed with one blocked worker and one occupied queue slot, then releasing both and delivering another message: ```text ORDERLY: nextDispatch=false, resumed=false, cached=2, poolQueue=0 ``` ### [P1] Keep retrying concurrent requests after a second rejection The [external executor is used directly](https://github.com/apache/rocketmq/blob/b3c92c9d7058cb953c7917c28757ec181e169ce6/client/src/main/java/org/apache/rocketmq/client/impl/consumer/AbstractConsumeMessageService.java#L34-L45), but the delayed retry in [Pull concurrent consumption](https://github.com/apache/rocketmq/blob/b3c92c9d7058cb953c7917c28757ec181e169ce6/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java#L305-L316) and [POP concurrent consumption](https://github.com/apache/rocketmq/blob/b3c92c9d7058cb953c7917c28757ec181e169ce6/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessagePopConcurrentlyService.java#L292-L303) calls `consumeExecutor.submit` without handling another rejection. The first rejection schedules a retry after five seconds. If the executor is still saturated, the second rejection terminates that scheduled task exceptionally; its Future is not observed and no further retry is scheduled. In Pull mode, messages remain in `ProcessQueue` and pin the offset. Since consumption never started, `cleanExpiredMsg` also cannot expire them using the consumption-start timestamp. In POP mode, `ConsumeRequest.run` never executes its timeout cleanup, leaving `waitAckCounter` inflated; broker redelivery does not release that client-side count. Please handle repeated rejection in both delayed callbacks, retaining retries while the consumer is active and handling dropped queues, shutdown, and POP expiration appropriately. Regression tests should keep the pool full across the first delayed retry, then release capacity and assert eventual consumption and offset/cache cleanup. For POP, also assert that the pending-ACK count is released when the message expires before execution. Observed after the second rejection and subsequent executor recovery: ```text CONCURRENT: resumed=false, retained=1, poolQueue=0 POP_CONCURRENT: resumed=false, retained=1, poolQueue=0 ``` ### Validation and review coverage - JDK 21: existing relevant client tests had 44 passes and 3 parameter-case skips; the three additional recovery regression tests failed as described above. Proxy executor/sharing/HeartbeatSyncer tests had 7 passes. JaCoCo was disabled, and Proxy tests required `-Dnet.bytebuddy.experimental=true` for the existing Byte Buddy version. Checkstyle/SpotBugs were not rerun. - Call sites and callbacks inspected: `DefaultMQPushConsumerImpl.start`, all four consume-service constructors and submit/run/result paths, Pull/POP success/error callbacks, delayed retries, orderly lock/continuation callbacks, expiration cleanup, pool resizing and running-info callers, consumer shutdown/persistence/rebalance cleanup, and `ClusterServiceManager -> ClusterConsumerManager -> HeartbeatSyncer -> AbstractSystemMessageSyncer` including listener registration and reverse shutdown order. - Configuration branches inspected: null versus injected executors; bounded/unbounded and virtual-thread executors; concurrent/orderly and broadcasting/clustering; `clientRebalance` on/off; Proxy local/cluster mode and default/custom pool sizes. POP orderly's existing incomplete implementation was not attributed to this PR. - Compatibility inspected statically: old/new client-server combinations, mixed versions, rolling upgrade/downgrade, restart, existing offset files and corrupt-file backup fallback. No wire or persisted-format change was found; no live mixed-version or recovery integration test was run. - Failure paths inspected: synchronous and asynchronous exceptions, null/failure listener results, repeated rejection, timeout, cancellation, interruption, retry, offset persistence, unlock and shutdown cleanup. Existing ownership tests submit ordinary executor tasks; they still lack an end-to-end consumer shutdown test with actual queued/in-flight messages and a surviving sibling consumer. I recommend addressing the two rejection-recovery paths before merging the public executor-injection API. -- 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]
