[Bug] RocketMQUtil.getResult leaks pending-future entries in the static response map on timeout/failure
- 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