apache / apache/rocketmq

[Bug] Proxy processor futures can remain pending when completion tasks are rejected

Open
#10,739 3 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
22.6k
Forks
12k
Avg merge
3d 1h
Merged PRs (30d)
27

Description

## RocketMQ version

`develop` at `00e45b8a6db23efbe756d0306f10716156cfd4dd`

## Describe the bug

`FutureUtils.appendNextFuture` ignores the dependent stage returned by
`whenCompleteAsync`. If the supplied executor rejects the completion task, that
dependent stage completes exceptionally but the manually created `nextFuture`
is never completed.

The production executors used by `ProducerProcessor` and `ConsumerProcessor`
add a second failure mode. They are created through the default
`ThreadPoolMonitor.createAndMonitor` overload, whose rejection handler is
`DiscardOldestPolicy`. After shutdown it silently discards the completion task;
under saturation it can silently discard an older queued completion task.
Neither case gives `FutureUtils` a rejection that it can propagate, so a proxy
request future can remain pending indefinitely.

## Deterministic reproduction

Two synchronous regression tests reproduce the complete failure chain on JDK 8.

### 1. Rejection is not propagated by `FutureUtils`

1. Create a single-thread executor and shut it down.
2. Call `FutureUtils.addExecutor(CompletableFuture.completedFuture("value"), executor)`.
3. Inspect the returned future.

Expected: the returned future is completed exceptionally with the scheduling
failure.

Actual: the stage returned by `whenCompleteAsync` is exceptional, but the stage
is ignored and the returned `nextFuture` remains pending.

The controlled test
`FutureUtilsTest#testAddExecutorCompletesExceptionallyWhenExecutorRejectsTask`
failed at the expected `result.isDone()` assertion in 5/5 runs.

### 2. The actual Proxy processor executor silently drops the task

1. Construct `DefaultMessagingProcessor` with its real producer and consumer
processor executors.
2. Shut down the producer processor executor.
3. Pass that executor to `FutureUtils.addExecutor` with an already completed
source future.
4. Inspect the returned future.

Expected: rejection is observable and the returned future is completed
exceptionally.

Actual: `DiscardOldestPolicy` silently discards the task after shutdown and the
returned future remains pending.

The controlled test
`DefaultMessagingProcessorTest#testProcessorFutureCompletesWhenExecutorIsShutDown`
failed at the expected `result.isDone()` assertion in 5/5 runs. It uses the
actual processor pool, no sleeps, randomized scheduling, network access, or
external service.

## Impact

`FutureUtils.addExecutor` is used by the Proxy `ProducerProcessor` and
`ConsumerProcessor` request paths. During processor-pool shutdown, or when the
bounded pool rejects completion work, callers may receive a future that never
reaches a terminal state. This can leave gRPC/remoting requests waiting until an
outer timeout or connection teardown instead of receiving an immediate failure.

## Suggested fix

Both parts are needed; either one alone leaves a pending-future path:

- Retain the stage returned by `whenCompleteAsync` and propagate a scheduling
failure into `nextFuture`.
- Create the producer and consumer processor executors with an explicit
throwing rejection policy (for example `AbortPolicy`) instead of the default
silent `DiscardOldestPolicy`, so `FutureUtils` can observe both shutdown and
overload rejection.
- Add regression coverage for both a throwing executor and the actual Proxy
processor executor after shutdown.

This changes only internal completion/rejection behavior and does not change a
public protocol.

## Related work checked

- #8900 contains a `FutureUtils` stack frame but reports client cancellation and
message redelivery, not executor rejection or a pending returned future.
- #9253 concerns receipt-handle renewal after client disconnection, not
completion-task rejection.
- #5575 introduced the Proxy remoting implementation and is not a fix for this
behavior.
- #10711 concerns unsupported metadata futures and does not modify
`FutureUtils` or the processor executor policies.

Searches across open and closed issues and pull requests for `FutureUtils`,
`addExecutor`, `DefaultMessagingProcessor`, `DiscardOldestPolicy`, executor
shutdown/rejection, and pending Proxy futures found no equivalent report or
claimed fix.

Contributor guide

Open the contributing guide

Research direction

Start with FutureUtils.addExecutor and the producer and consumer processor executor creation in DefaultMessagingProcessor, then read the named regression tests: FutureUtilsTest#testAddExecutorCompletesExceptionallyWhenExecutorRejectsTask and DefaultMessagingProcessorTest#testProcessorFutureCompletesWhenExecutorIsShutDown. Run both tests and verify that rejected completion work completes the returned future exceptionally, including after processor executor shutdown.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
backend, distributed-systems, testing-qa
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
68/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.