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 08:23:02 UTC
(eventmesh) branch develop updated: Revert #5330, #5331, #5333, #5334 (prepare consolidated re-land with both contributor trailers) (#5335)
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 d9c90899e Revert #5330, #5331, #5333, #5334 (prepare consolidated re-land with both contributor trailers) (#5335)
d9c90899e is described below
commit d9c90899e0fd4ee8c179f74f6e58a7edf863f160
Author: Eason Chen <qq...@gmail.com>
AuthorDate: Mon Sep 7 16:22:56 2026 +0800
Revert #5330, #5331, #5333, #5334 (prepare consolidated re-land with both contributor trailers) (#5335)
Reverts the four commits landed earlier to attribute / merge #5316 (zhang-arvin) and #5325 (wangyusheng1985) onto develop:
- #5330 (c7de75b) — barrier fix for the RocketMQ 5 POP broker ACK (`fix(#5295): gate RocketMQ 5 POP broker ACK on distribution completion`)
- #5331 (4c3e85b) — README Quick start anchors 0-diff
- #5333 (2d2f98f) — source-level attribution Javadoc in `UniIngressService.java`
- #5334 (e6e12474) — package-level doc in `ingress/package-info.java`
After this revert lands, the develop tree returns to `2b0d7abb` (the pre-#5330 state), with the four changes unsuperseded.
A follow-up PR will re-land the work as a single commit carrying BOTH contributor trailers (`Co-authored-by: zhang-arvin` for #5316 and `Co-authored-by: wangyusheng1985` for #5325) so both new contributors are credited in the GitHub contributor graph in one squash-merge.
Why revert first? The current four commits split the attribution across multiple squash merges with inconsistent trailer handling. Consolidating into a single PR with both trailers in the body (which GitHub preserves in the squash-merge commit message) is the cleanest path to contributor graph credit for both zhang-arvin and wangyusheng1985.
---
.../runtime/ingress/UniIngressService.java | 26 +++++-----------------
.../eventmesh/runtime/ingress/package-info.java | 10 ---------
2 files changed, 5 insertions(+), 31 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 4f34e0257..f6e617abb 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
@@ -372,29 +372,13 @@ public class UniIngressService {
// Frame architecture: the POP check key rides in frame attributes (empopck) and the
// deferred broker ACK fires on client ACK — same at-least-once goal, Frame-native.
long offset = nextOffset(topic);
- // P2 fix (PR #5316 by zhang-arvin, fixes #5295): 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).
+ // 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");
- 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);
- }
+ 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);
}
}
UniTrace.end(dispatchSpan);
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/package-info.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/package-info.java
index a903e6ba8..1d267445b 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/package-info.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/package-info.java
@@ -22,16 +22,6 @@
*
* <p>Policy: Public facade -- boot wires ingress into UniRuntime, but no engine sub-package may bypass it.
*
- * <p>Broker-ACK barrier (RocketMQ 5.x POP, see PR #5316 by zhang-arvin,
- * fixes #5295): when the ingress frame carries a POP check key (the
- * {@code empopck} attribute), deliveries to all matched subscriptions
- * share an {@link java.util.concurrent.atomic.AtomicInteger} counter
- * initialized to the target count; the broker is ACKed (via
- * {@code storage.ackPulledMessage}) only when the last required delivery
- * ACKs. This restores at-least-once semantics across LOAD_BALANCE,
- * BROADCAST, and MULTICAST distribution modes. Frames without
- * {@code empopck} bypass the barrier (no broker ACK to defer).
- *
* <p>Marked {@link org.apache.eventmesh.common.Internal @Internal} as a
* whole package; public types must carry {@link org.apache.eventmesh.common.Public @Public}.
*/
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org