apache / apache/rocketmq-a2a

[Bug] RocketMQUtil.getResult leaks pending-future entries in the static response map on timeout/failure

Open
#40 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
47
Forks
16
Avg merge
50m
Merged PRs (30d)
1

Description

### Describe the Bug

`RocketMQUtil.getResult` registers a `(msgId -> A2AResponseFuture)` entry in the static `MESSAGE_RESPONSE_MAP`, then blocks on `future.get(120, SECONDS)` and only removes the entry **after** `get` returns normally:

```java
msgIdAndAsyncTypedMap.put(responseMessageId, new A2AResponseFuture(completableFuture, typeReference));
String result = completableFuture.get(120, TimeUnit.SECONDS); // may throw
msgIdAndAsyncTypedMap.remove(responseMessageId); // skipped on any exception
return result;
```

`get` can throw `TimeoutException` (no reply within 120s), `ExecutionException` (failed reply), or `InterruptedException` — in all three cases the `remove` line is never reached. The completion side (`processNonStreamResult`) only calls `complete(...)` and never removes, and there is no periodic sweeper in the class, so the successful path of `getResult` is the only cleanup point.

Every timed-out or failed request therefore leaks one entry in a class-level static map for the lifetime of the JVM. Since tolerating slow/unavailable remote agents is exactly the scenario this transport exists for, the leak accumulates steadily and eventually OOMs a long-running process.

### Steps to Reproduce

Deterministic test (included in the incoming PR): register via a background `getResult`, then `completeExceptionally` the future — the entry stays in `MESSAGE_RESPONSE_MAP` on the unfixed code.

### What Did You Expect to See?

The pending entry is removed on every exit path of `getResult`.

### What Did You See Instead?

Entries leak on timeout/failure/interrupt and the static map grows without bound.

### Additional Context

Fix incoming: wrap the `get` in `try { ... } finally { msgIdAndAsyncTypedMap.remove(responseMessageId); }`. The streaming map (`MESSAGE_STREAM_RESPONSE_MAP`) may deserve the same audit as a follow-up.

Contributor guide

No contributing guide indexed for this repository

Research direction

Start at RocketMQUtil.getResult and inspect how entries are added to and removed from MESSAGE_RESPONSE_MAP around the timed future.get call. Reproduce the failure by completing the future exceptionally from a background getResult call; done means timeout, failure, and interruption paths all remove the pending entry, while the successful path still works.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
distributed-systems
Issue type
Bug
Difficulty
2/5
Estimated time
1-3 hours
Activity status
Active
Clarity
Clearly specified
Newbie friendliness
58/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.