[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)

[qqeasonchen]

@zhang-arvin please fix the conflicts and review blockers,thx

[qqeasonchen]

@zhang-arvin 谢谢原 PR！本意保留你的实现，但 #5316 基于 pre-#5301 的 903 行 UniIngressService.java 整体重写，跟 develop HEAD（988 行，已被 #5311 / #5312 重构过）有结构性冲突——3-way merge 会产生几十处冲突标记而不是几处。

所以我重做了 rebase：保留你设计的 barrier 逻辑（AtomicInteger pending = new AtomicInteger(targets.size()) + decrementAndGet() == 0 触发 broker ACK），把它移植到 develop HEAD 当前的代码结构上（保留 TTL 检查、Otel span、nextOffset、UniTrace.end、subscriptionManager.targetsFor 复用）。./gradlew :eventmesh-runtime:compileJava 通过。

新 PR: #5330（merge-5316-pop-broker-ack 分支）。

这个 #5316 因为 base 已经完全过时，建议直接关闭。如果你想在自己的 PR 上发，请以 #5330 为底 cherry-pick —— 我会重命名 #5330 branch 为 fix/5295-pop-broker-ack-barrier、把 author 改成你（这样 git history 显示是你提的），然后你 reopen 即可。

Fixes #5295 的功劳归你。

[qqeasonchen]

Closing as superseded. The barrier fix from this PR was conflict-resolved and landed in #5330 (commit c7de75b); the contributor attribution for zhang-arvin is being recorded via the companion PR #5333, which adds an attribution Javadoc and carries a Co-authored-by: zhang-arvin <arvin.zhang@htx-inc.com> trailer in the commit message (same pattern that #5331 used for #5325 / wangyusheng1985).

Thanks for the contribution @zhang-arvin — your original commit 1cdca51f is the canonical reference for the broker-ACK barrier logic.

[qqeasonchen]

Attribution update (post history-cleanup): the close comment above references c7de75b (#5330) and #5333/#5334, which were later consolidated during a develop history cleanup. The final landing for this PR's contribution is now:

• Commit 5fb1f6b18eaf983940b9313061abce3463d5f3b5 on develop — carries Co-authored-by: zhang-arvin <arvin.zhang@htx-inc.com> in the commit message, and contains the broker-ACK barrier code in UniIngressService.java (per-frame AtomicInteger + per-delivery decrement + late broker ack) plus the package-level barrier doc in ingress/package-info.java.

zhang-arvin is credited as co-author on that commit, so this PR's contribution is reflected in the GitHub contributor graph via the trailer. Thanks again @zhang-arvin!