qqeasonchen opened a new pull request, #5330:
URL: https://github.com/apache/eventmesh/pull/5330
## What changes were proposed in this pull request
Fix #5295: Gate RocketMQ 5 POP broker ACK on distribution completion.
### Problem
Previously, a single `mqAck` callback was shared across all deliveries of a
frame. In BROADCAST/MULTICAST mode, the first client ACK would immediately ACK
the broker, even if other required targets had not yet received or acknowledged
the message.
### Solution
Introduce a broker-ACK barrier using an `AtomicInteger` counter:
- All deliveries of the same frame share a single counter initialized to
`targets.size()`
- Broker ACK fires only when all deliveries have ACKed (counter reaches 0)
- LOAD_BALANCE (1 target): 1 ACK → broker ACK
- BROADCAST (N targets): N ACKs → broker ACK
- MULTICAST (matched targets): all matched ACKs → broker ACK
### Notes on this rebase
The original PR (#5316, zhang-arvin) was based on the pre-#5301
`UniIngressService.java` (903 lines) and conflicted wholesale with the
post-#5301 file (988 lines, after Sub-PR B #5311 + Sub-PR C #5312). This PR
applies **the same barrier fix** to the current develop file:
- `String popCk = f.attributes().get("empopck")` block is rewritten from a
single shared `mqAck` lambda to the barrier pattern.
- The TTL check (`isExpired(f)`), Otel span, `nextOffset(topic)`, and
`UniTrace.end(dispatchSpan)` from develop HEAD are preserved.
- `List<Subscription> targets = subscriptionManager.targetsFor(topic, f)` is
computed once and reused for both branches (no extra `subscriptionManager`
call).
- No `popCk` path: unchanged (null callback, no broker ack).
### Changes
- `eventmesh-runtime/.../UniIngressService.java`: Replace the shared `mqAck`
callback with a per-frame `AtomicInteger` barrier; broker ACK only fires when
the last required delivery ACKs.
### Verification
- [x] `./gradlew.bat :eventmesh-runtime:compileJava` — BUILD SUCCESSFUL
(only deprecation warnings unrelated to this change)
- [ ] In BROADCAST mode, the first of multiple client ACKs does not ACK the
POP message at the broker
- [ ] The final required ACK executes exactly one broker ACK
- [ ] Duplicate and out-of-order ACKs do not execute multiple broker ACKs
(guarded by `ReliableDispatcher.ack`'s idempotency)
- [ ] Runtime failure before completion causes broker redelivery after POP
invisible time
### Credit
Original implementation by @zhang-arvin in #5316. The barrier logic, control
flow, and intent are unchanged; this PR re-applies them to the current
`develop` branch's `UniIngressService.java`.
Fixes #5295
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
