[ISSUE #11172] Respond when async send storage future fails

[zmuxuny]

Which Issue(s) This PR Fixes

Fixes #11172

Brief Description

With asyncSendEnable=true, SendMessageProcessor returns null before the store operation completes and relies on the store future callback to write the broker response. Both the single-message and batch paths used thenAcceptAsync, so an exceptionally completed store future skipped the callback entirely. The client then received no broker response and could only terminate through an outer request timeout or connection teardown.

This change handles both completion outcomes for single and batch async sends:

• normal PutMessageResult values keep the existing response/status/metrics behavior;
• exceptional completion logs the original throwable on the broker and immediately responds with SYSTEM_ERROR using a generic client-safe remark;
• the existing after-send hook still runs once for the exceptional path;
• synchronous send behavior and wire/storage formats are unchanged.

Fail-before evidence

Baseline: develop at bc33e8e4d7b25089af5f51bc669bdfedabfebe7d, Amazon Corretto 8u432.

A deterministic regression makes MessageStore.asyncPutMessage(...) return a future already completed exceptionally, invokes a normal SEND_MESSAGE, and waits one second for channel.writeAndFlush.

On the unmodified baseline:

Tests run: 1, Failures: 0, Errors: 1
ConditionTimeoutException: response was not fulfilled within 1 seconds

The preceding 10 modules in the broker -am reactor all succeeded. The request processor had returned null, but no response was ever written.

How Did You Test This Change?

Focused exceptional-completion regressions cover both paths:

• single-message asyncPutMessage failure returns SYSTEM_ERROR instead of leaving the request pending;
• batch asyncPutMessages failure has the same behavior.

Final verification command:

mvn -B -ntp -pl broker -am \
  -Dtest=SendMessageProcessorTest \
  -Dsurefire.failIfNoSpecifiedTests=false test

Result: 18 tests passed, 0 failures/errors/skips in SendMessageProcessorTest. All 11 modules in the broker -am reactor completed successfully. Checkstyle reported 0 violations in every checked module, and SpotBugs reported 0 bug instances / 0 errors in every reactor module where the project runs SpotBugs (the generated rocketmq-proto module uses its existing project-configured skip).

git diff --check also passes.

[RockteMQ-AI]

Summary

This PR fixes a real bug where clients hang indefinitely when async storage fails. The change from thenAcceptAsync to whenCompleteAsync correctly handles both normal and exceptional completion paths for single-message and batch async sends.

Review

• Correctness ✅ — The exception handler properly sends SYSTEM_ERROR response to clients, invokes the after-send callback, and avoids leaking internal error details
• Performance ✅ — No performance concerns; same execution model with additional error handling
• Tests ✅ — Two new tests cover both single-message and batch async failure scenarios with proper async verification
• Compatibility ✅ — No API/wire format changes; synchronous send path unchanged

Minor Observation

The error path logs the exception and responds to the client, but does not emit broker-side metrics for the failure (the normal path goes through handlePutMessageResult which likely handles metrics). This may be intentional since the exception itself is logged, but worth considering if monitoring dashboards should track async storage failures separately.

Overall: Clean, focused fix with excellent PR description and test coverage. LGTM.

---

Automated review by github-manager-bot