You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by ch...@apache.org on 2026/09/07 07:15:59 UTC
(eventmesh) branch develop updated: fix(#5295): gate RocketMQ 5 POP broker ACK on distribution completion (merged from PR #5316) (#5330)
This is an automated email from the ASF dual-hosted git repository.
qqeasonchen pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/develop by this push:
new c7de75b39 fix(#5295): gate RocketMQ 5 POP broker ACK on distribution completion (merged from PR #5316) (#5330)
c7de75b39 is described below
commit c7de75b39f6693dbb472104d273f48eab815a1af
Author: Eason Chen <qq...@gmail.com>
AuthorDate: Mon Sep 7 15:15:53 2026 +0800
fix(#5295): gate RocketMQ 5 POP broker ACK on distribution completion (merged from PR #5316) (#5330)
Cherry-pick of zhang-arvin's barrier change, rebased on develop HEAD after the
#5301 Sub-PR B (#5311) and Sub-PR C (#5312) refactors that rewrote the
UniIngressService deliver path. The original PR was based on the pre-#5301 file
(903 lines) and conflicted wholesale with the post-#5301 file (988 lines); this
commit applies the same barrier fix (AtomicInteger + per-delivery decrement + late
broker ack) to the current develop file structure.
- Shared mqAck (first ACK = broker ACK, premature for BROADCAST/MULTICAST)
-> per-frame AtomicInteger initialized to targets.size(); broker ack only
fires when the last required delivery ACKs (counter reaches 0).
- LOAD_BALANCE (1 target): 1 ACK -> broker ACK (semantic unchanged).
- BROADCAST (N targets): N ACKs -> broker ACK.
- MULTICAST (matched targets): all matched ACKs -> broker ACK.
- No popCk path: unchanged (null callback, no broker ack).
- Reuses SubscriptionManager.targetsFor(); preserves existing TTL check,
Otel span, metrics, and Frame-architecture context from develop HEAD.
Original PR: https://github.com/apache/eventmesh/pull/5316 (zhang-arvin)
Fixes #5295
---
.../runtime/ingress/UniIngressService.java | 22 ++++++++++++++++++----
1 file changed, 18 insertions(+), 4 deletions(-)
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/UniIngressService.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/UniIngressService.java
index f6e617abb..a93b63bd8 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/UniIngressService.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/UniIngressService.java
@@ -375,10 +375,24 @@ public class UniIngressService {
// P2 fix: if the frame carries a POP check key (RocketMQ 5.x deferred ACK), build a
// callback that ACKs the broker on client ACK (restoring at-least-once).
String popCk = f.attributes().get("empopck");
- Runnable mqAck = (popCk != null) ? () -> storage.ackPulledMessage(topic, popCk) : null;
- for (Subscription target : subscriptionManager.targetsFor(topic, f)) {
- dispatcher.deliver(topic, partition, offset, f, target.getClientId(),
- channelFor(target.getClientId()), mqAck);
+ List<Subscription> targets = subscriptionManager.targetsFor(topic, f);
+ if (popCk != null && !targets.isEmpty()) {
+ java.util.concurrent.atomic.AtomicInteger pending =
+ new java.util.concurrent.atomic.AtomicInteger(targets.size());
+ Runnable mqAck = () -> {
+ if (pending.decrementAndGet() == 0) {
+ storage.ackPulledMessage(topic, popCk);
+ }
+ };
+ for (Subscription target : targets) {
+ dispatcher.deliver(topic, partition, offset, f, target.getClientId(),
+ channelFor(target.getClientId()), mqAck);
+ }
+ } else {
+ for (Subscription target : targets) {
+ dispatcher.deliver(topic, partition, offset, f, target.getClientId(),
+ channelFor(target.getClientId()), null);
+ }
}
}
UniTrace.end(dispatchSpan);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org