[Fix] Gate RocketMQ 5 POP broker ACK on distribution completion (#5295)

[zhang-arvin]

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

Changes

• eventmesh-runtime/.../UniIngressService.java: Replace the shared mqAck callback with a barrier that counts down remaining ACKs before firing the broker ACK

Verification

• 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

[github-actions]

Welcome to the Apache EventMesh community!!
This is your first PR in our project. We're very excited to have you onboard contributing. Your contributions are greatly appreciated!

Please make sure that the changes are covered by tests.
We will be here shortly.
Let us know if you need any help!

Want to get closer to the community?

WeChat Assistant	WeChat Public Account	Slack
		Join Slack Chat

Mailing Lists:

Name	Description	Subscribe	Unsubscribe	Archive
Users	User support and questions mailing list	Subscribe	Unsubscribe	Mail Archives
Development	Development related discussions	Subscribe	Unsubscribe	Mail Archives
Commits	All commits to repositories	Subscribe	Unsubscribe	Mail Archives
Issues	Issues or PRs comments and reviews	Subscribe	Unsubscribe	Mail Archives

[qqeasonchen]

Review: changes requested before merge

Thanks for the fix — the barrier approach is correct and the code is clean. However, there are 3 blockers and 3 suggestions that need to be addressed.

What I did

1. Fetched the PR head (1cdca51) into a local pr-5316 ref
2. Pulled both blobs via git show and ran diff -u locally
3. Key finding: the file changed from CRLF to LF, which inflates the diff to +920/-903; the actual logic change is only ~20 lines

---

🔴 Blocker 1: Missing tests

UniIngressService is on the hot path (every message goes through it), and this PR changes the condition that fires the broker ACK — a premature trigger loses messages, a delayed trigger leaks memory / causes duplicate consumption. But the PR has no test file changes.

Issue #5295's acceptance criteria explicitly requires:

• Tests cover BROADCAST, LOAD_BALANCE, and MULTICAST completion rules.
• A RocketMQ 5 broker E2E test verifies this behavior.

The 4 verification checkboxes in the PR description are all unchecked (PMC convention requires author self-verification first).

Minimum required (any of these will do):

• Unit test (add UniIngressServiceTest or extend ReliableDispatcherTest): mock MeshStoragePlugin + mock ReliableDispatcher, verify:
  • LOAD_BALANCE (1 target): 1 client ACK → 1 ackPulledMessage trigger
  • BROADCAST (3 targets): all 3 client ACKs → 1 trigger; intermediate ACKs do not trigger
  • MULTICAST (2 matched): all 2 ACKs → 1 trigger
  • Duplicate ACK (same deliveryId): counter goes negative, broker ACK is not re-triggered
  • popCk == null: else branch passes null mqAck (equivalent to no-barrier behavior)
• In-process E2E (in the style of ClusterDeliveryFaultTest): use the existing InMemoryMetaStore + real UniIngressService + mock storage, run at least the full BROADCAST barrier flow

For reference, see PR #5308 (the #5293 implementation) which added 5 ClusterDeliveryFaultTest scenarios in the same style.

🔴 Blocker 2: Barrier duplicate-ACK protection is incomplete

Runnable mqAck = () -> {
    if (pending.decrementAndGet() == 0) {  // ← issue here
        storage.ackPulledMessage(topic, popCk);
    }
};

Problem: decrementAndGet == 0 only fires on the first time it reaches zero. But there are 3 scenarios that cause the mqAck to be entered more than expected:

1. Same clientId retries ACK (SDK-side retry / network resend): ReliableDispatcher.ack() should be idempotent, but even if it is, the barrier will continue to decrement
2. ACK for a non-matching deliveryId (potentially introduced in the future): decrements a counter that shouldn't be included in the barrier
3. Re-dispatch of the same frame (forward path / requeue)

Note: in repeat-ACK scenarios, decrementAndGet == 0 is only true the first time, and subsequent -1, -2... won't re-trigger — this is actually OK in isolation. But there is one real risk:

If ReliableDispatcher.ack is not strictly idempotent (per the issue #5295 description it dedupes, but if any race is missed), the storage.ackPulledMessage call outside the barrier could race. Recommend an explicit CAS guard:

AtomicInteger pending = new AtomicInteger(targets.size());
AtomicBoolean brokerAcked = new AtomicBoolean(false);
Runnable mqAck = () -> {
    if (pending.decrementAndGet() == 0 && brokerAcked.compareAndSet(false, true)) {
        storage.ackPulledMessage(topic, popCk);
    }
};

This way, even if decrementAndGet somehow reaches 0 multiple times (theoretically impossible but defensive), the broker ACK is only triggered once.

🔴 Blocker 3: multi-instance path not handled

if (cluster != null) {
    // Multi-instance: route via the cluster coordinator (local vs cross-instance forward).
    cluster.dispatch(topic, f);
} else {
    // barrier logic
}

The PR only fixes the else branch. But cluster.dispatch internally still calls storage.poll → its own deliver loop — the same bug will reproduce in the multi-instance path.

Please confirm whether cluster.dispatch internally also goes through storage.poll + the same target-allocation logic; if so, the barrier must be added there as well (or refactored into a shared helper).

---

🟡 Suggestion 1: Split the file-format normalization into a separate PR

99% of the +920/-903 diff is CRLF → LF noise (each line loses 1 byte → line count grows; plus the actual +17 line logic change). Either normalize the file standalone first (LF across all .java, or keep CRLF if the repo default is CRLF), or use .gitattributes to keep the line-ending story consistent.

🟡 Suggestion 2: The null mqAck else branch can be cleaner

The current else branch explicitly loops with null mqAck. Consider:

if (popCk != null && !targets.isEmpty()) {
    // ... barrier setup
} else {
    for (Subscription target : targets) {
        dispatcher.deliver(..., null);  // can be simplified if dispatcher tolerates null
    }
}

But this requires confirming ReliableDispatcher.deliver accepts a null mqAck — the existing code already passes null, so the current state is acceptable.

🟡 Suggestion 3: Annotate the issue reference and scope

The code added // Issue #5295: comment, which is good. Recommend also adding:

// Issue #5295: the barrier is per-frame. The multi-instance cluster.dispatch path
// is out of scope for this PR (see PR review comment #3) and will be tracked
// separately.

This makes the multi-instance path's handling state explicit so it isn't mistaken for fixed in the future.

---

Summary

Category	Item	Status
Blocker	1. Missing tests	Must add
Blocker	2. Barrier duplicate-ACK protection incomplete	Add compareAndSet guard
Blocker	3. multi-instance path not covered	Must clarify or fix
Suggestion	1. Split file-format normalization PR	Optional
Suggestion	2. null mqAck else branch	Optional
Suggestion	3. Annotate issue reference	Recommended

Ping me for re-review after the blockers are addressed.

— qqeasonchen (apache/eventmesh PMC)