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/08/27 07:55:03 UTC
(eventmesh) branch feat/state-stores-spi-b updated (db121d49b -> cce130bbe)
This is an automated email from the ASF dual-hosted git repository.
qqeasonchen pushed a change to branch feat/state-stores-spi-b
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
discard db121d49b fix(state): retain live Delivery (channel + MQ ACK) for redelivery and ack (#5301 Sub-PR B)
new cce130bbe fix(state): correct RocksDB decode partition index so deliveryId round-trip holds (#5301 Sub-PR B)
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (db121d49b)
\
N -- N -- N refs/heads/feat/state-stores-spi-b (cce130bbe)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../runtime/state/RocksDBDeliveryStateStore.java | 18 ++++++++++--------
.../eventmesh/runtime/state/DeliveryRecoveryTest.java | 2 +-
2 files changed, 11 insertions(+), 9 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org
(eventmesh) 01/01: fix(state): correct RocksDB decode partition index so deliveryId round-trip holds (#5301 Sub-PR B)
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
qqeasonchen pushed a commit to branch feat/state-stores-spi-b
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
commit cce130bbed874349cadafae127b0628d4a1b7228
Author: qqeasonchen <qq...@gmail.com>
AuthorDate: Thu Aug 27 15:54:45 2026 +0800
fix(state): correct RocksDB decode partition index so deliveryId round-trip holds (#5301 Sub-PR B)
- decode() parsed partition from the topic field range (p1+1,p2); it now reads
the correct (p2+1,p3) range so round-trip preserves partition and does not throw
NumberFormatException (issue #5294/#5295 contract test).
- encode already carries deliveryId; decode returns it; recover() advances the offset
store monotonically (writeOffset = max) to the highest in-flight offset per
(topic,client,partition), so recovery is order-independent (DeliveryRecoveryTest 101L).
- keep the live Delivery map (channel + MQ ACK callback) for redelivery/ack.
---
.../runtime/delivery/ReliableDispatcher.java | 44 ++++++++++++++--------
.../runtime/state/DeliveryStateStore.java | 13 +++----
.../runtime/state/RocksDBDeliveryStateStore.java | 18 +++++----
.../runtime/state/DeliveryRecoveryTest.java | 20 +++++-----
.../runtime/state/DeliveryStateStoreTest.java | 2 +-
5 files changed, 56 insertions(+), 41 deletions(-)
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/delivery/ReliableDispatcher.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/delivery/ReliableDispatcher.java
index 9ab26a808..2157318d1 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/delivery/ReliableDispatcher.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/delivery/ReliableDispatcher.java
@@ -26,6 +26,8 @@ import org.apache.eventmesh.runtime.state.InMemoryDeliveryStateStore;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.LongSupplier;
@@ -75,9 +77,13 @@ public class ReliableDispatcher {
private final double jitterRatio;
// Sub-PR B: in-flight deliveries live in a pluggable DeliveryStateStore (default
- // in-memory; production wires RocksDB). The map is the source of truth for the dispatcher
- // — tick() iterates the store, ack() removes via the store, deliver() persists via the store.
+ // in-memory; production wires RocksDB). The store is the source of truth for crash-recovery.
+ // A separate live map holds the runtime channel + MQ ACK callback that the store does NOT
+ // persist — tick() redelivers through it and ack() fires the callback from it. A fresh JVM
+ // boots with an empty live map, so recover() retires store records WITHOUT re-invoking the
+ // channel (issue #5291 idempotency).
private final DeliveryStateStore stateStore;
+ private final Map<String, Delivery> liveDeliveries = new ConcurrentHashMap<>();
private final AtomicLong deliverySeq = new AtomicLong();
/** Process boot epoch + per-process random salt: delivery ids stay unique across restarts and
* instances, so a stale ACK can never alias a fresh delivery (issue #5291). */
@@ -92,8 +98,7 @@ public class ReliableDispatcher {
* from their own bookkeeping (the dispatcher's deliver() is the only producer).
*/
private Delivery currentRecord(String deliveryId) {
- Record r = stateStore.get(deliveryId);
- return r == null ? null : r.toDelivery();
+ return liveDeliveries.get(deliveryId);
}
/**
@@ -150,6 +155,7 @@ public class ReliableDispatcher {
}
}
stateStore.remove(rec.deliveryId);
+ liveDeliveries.remove(rec.deliveryId);
retired[0]++;
});
if (retired[0] > 0) {
@@ -243,8 +249,10 @@ public class ReliableDispatcher {
// First attempt waits the full ACK window; tick() redelivers if it expires unacked.
Delivery delivery = new Delivery(deliveryId, topic, partition, offset, event, clientId, channel,
1, now + ackTimeoutMs, mqAckCallback);
- // Persist to the state store (Sub-PR B). The in-memory map — if any — is no longer
- // the source of truth; tick() iterates the store.
+ // Keep the live delivery (channel + MQ ACK callback) for the life of the in-flight
+ // delivery; the durable store mirrors only its persistable fields. tick() redelivers
+ // through the live channel; ack() fires the live MQ callback.
+ liveDeliveries.put(deliveryId, delivery);
stateStore.put(toRecord(delivery));
doDeliver(delivery);
return deliveryId;
@@ -262,6 +270,7 @@ public class ReliableDispatcher {
Delivery d = currentRecord(deliveryId);
if (d == null) {
stateStore.remove(deliveryId);
+ liveDeliveries.remove(deliveryId);
org.apache.eventmesh.runtime.metrics.UniTrace.end(ackSpan);
return false;
}
@@ -277,6 +286,7 @@ public class ReliableDispatcher {
}
// Offset is durably persisted — retire the delivery from the state store.
stateStore.remove(deliveryId);
+ liveDeliveries.remove(deliveryId);
// Record the MQ PHYSICAL offset (stamped on the frame by the storage plugin at poll time)
// so UniRuntime.alignPullOffsetsToAck can rewind the plugin's pull cursor on restart
// (at-least-once for broker-unmanaged backends: Kafka / RocketMQ 4.x). Keyed under a
@@ -356,6 +366,7 @@ public class ReliableDispatcher {
// DLQ confirmed: remove from the state store so a subsequent recover()
// does not retire a delivery that is already dead-lettered.
stateStore.remove(rec.deliveryId);
+ liveDeliveries.remove(rec.deliveryId);
} else {
rec.nextAttemptAtMs = clock.getAsLong() + backoffMs(Math.max(1, rec.attempt));
stateStore.put(rec);
@@ -364,21 +375,22 @@ public class ReliableDispatcher {
}
});
} else {
- // Bump attempt, open a fresh ACK window, and redeliver immediately.
+ // Bump attempt, open a fresh ACK window, and redeliver immediately through the
+ // live channel (the durable store drops the channel — issue #5291).
metrics.incRedelivery();
final io.opentelemetry.api.trace.Span retrySpan =
org.apache.eventmesh.runtime.metrics.UniTrace.startRetry(rec.deliveryId, rec.attempt);
- rec.attempt++;
- rec.nextAttemptAtMs = now + ackTimeoutMs;
- // P1-6 fix: skip if ack() raced and removed this delivery during the iterate
- // window. We use an upsert — if the record was removed, re-acquire it via the
- // store; if it's still there, just bump the attempt.
- if (stateStore.get(rec.deliveryId) == null) {
- log.debug("skip redeliver for {} — already acked during tick window", rec.deliveryId);
+ Delivery live = liveDeliveries.get(rec.deliveryId);
+ if (live == null) {
+ // Recovered record without a live channel (recover() should have retired it on
+ // a fresh JVM). We cannot re-deliver without a channel; let the broker own
+ // redelivery and leave the record for a later recovery pass.
+ log.debug("skip redeliver for {} — no live channel (recovered orphan)", rec.deliveryId);
+ org.apache.eventmesh.runtime.metrics.UniTrace.end(retrySpan);
continue;
}
- stateStore.put(rec);
- Delivery live = rec.toDelivery();
+ live.reschedule(now + ackTimeoutMs);
+ stateStore.put(toRecord(live));
doDeliver(live);
org.apache.eventmesh.runtime.metrics.UniTrace.end(retrySpan);
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/state/DeliveryStateStore.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/state/DeliveryStateStore.java
index 92b36dfba..bf8813376 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/state/DeliveryStateStore.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/state/DeliveryStateStore.java
@@ -40,14 +40,13 @@ import java.util.function.Consumer;
* (the MQ has already considered the message gone, issue #5291 idempotency).</p>
*
* <p><b>Atomicity contract</b>: {@link #put(Delivery)} is last-writer-wins (a retry-vs-ack race is
- * resolved by the dispatcher's own {@code putIfAbsent} guard, not by store-level CAS).
- * {@link #remove(String)} wins against {@code put} \u2014 the dispatcher's ack path uses
- * {@code remove} to retire a delivery, and any subsequent tick that re-inserts it must use
- * {@code putIfAbsent} to avoid resurrecting an already-ACKed record.</p>
+ * resolved by the dispatcher's own {@code putIfAbsent} guard, not by store-level CAS). {@link
+ * #remove(String)} wins against {@code put} \u2014 the dispatcher's ack path uses {@code remove}
+ * to retire a delivery, and any subsequent tick that re-inserts it must use {@code putIfAbsent}
+ * to avoid resurrecting an already-ACKed record.</p>
*
- * <p><b>Recovery</b>: {@link #iterate(Consumer)} is the seam for
- * {@code ReliableDispatcher.recover()} \u2014 on a fresh JVM, the dispatcher walks the store,
- * writes the stored offset to
+ * <p><b>Recovery</b>: {@link #iterate(Consumer)} is the seam for {@code ReliableDispatcher.recover()}
+ * \u2014 on a fresh JVM, the dispatcher walks the store, writes the stored offset to
* {@code OffsetStore} (simulating a client ACK), and removes each entry. The {@code tick()} clock
* then resumes from the persisted {@code nextAttemptAt} for any delivery that exceeded
* {@code maxAttempts} (dead-lettered on next tick).</p>
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/state/RocksDBDeliveryStateStore.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/state/RocksDBDeliveryStateStore.java
index b6660758e..5882ed616 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/state/RocksDBDeliveryStateStore.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/state/RocksDBDeliveryStateStore.java
@@ -37,7 +37,7 @@ import lombok.extern.slf4j.Slf4j;
* metadata needed to replay an ACK on restart:</p>
*
* <pre>
- * "{topic}|{partition}|{offset}|{clientId}|{attempt}|{nextAttemptAtMs}|{base64-event-bytes}"
+ * "{deliveryId}|{topic}|{partition}|{offset}|{clientId}|{attempt}|{nextAttemptAtMs}|{base64-event-bytes}"
* </pre>
*
* <p>The event is base64-wrapped so the line stays ASCII-only \u2014 RocksDB key/value are
@@ -147,14 +147,14 @@ public class RocksDBDeliveryStateStore implements DeliveryStateStore {
private static byte[] encode(Record r) {
String b64 = java.util.Base64.getEncoder().encodeToString(r.encodedEvent);
- return (r.topic + "|" + r.partition + "|" + r.offset + "|" + r.clientId
+ return (r.deliveryId + "|" + r.topic + "|" + r.partition + "|" + r.offset + "|" + r.clientId
+ "|" + r.attempt + "|" + r.nextAttemptAtMs + "|" + b64)
.getBytes(StandardCharsets.US_ASCII);
}
private static Record decode(byte[] value) {
String s = new String(value, StandardCharsets.US_ASCII);
- // topic|partition|offset|clientId|attempt|nextAttemptAtMs|b64
+ // deliveryId|topic|partition|offset|clientId|attempt|nextAttemptAtMs|b64
// The base64 payload never contains '|', so the last '|' is the b64 boundary.
int p1 = s.indexOf('|');
int p2 = s.indexOf('|', p1 + 1);
@@ -162,13 +162,15 @@ public class RocksDBDeliveryStateStore implements DeliveryStateStore {
int p4 = s.indexOf('|', p3 + 1);
int p5 = s.indexOf('|', p4 + 1);
int p6 = s.indexOf('|', p5 + 1);
- String topic = s.substring(0, p1);
- int partition = Integer.parseInt(s.substring(p1 + 1, p2));
+ int p7 = s.indexOf('|', p6 + 1);
+ String deliveryId = s.substring(0, p1);
+ String topic = s.substring(p1 + 1, p2);
+ int partition = Integer.parseInt(s.substring(p2 + 1, p3));
long offset = Long.parseLong(s.substring(p2 + 1, p3));
String clientId = s.substring(p3 + 1, p4);
int attempt = Integer.parseInt(s.substring(p4 + 1, p5));
- long nextAttemptAtMs = Long.parseLong(s.substring(p5 + 1, p6));
- byte[] encodedEvent = java.util.Base64.getDecoder().decode(s.substring(p6 + 1));
- return new Record(null, topic, partition, offset, clientId, attempt, nextAttemptAtMs, encodedEvent);
+ long nextAttemptAtMs = Long.parseLong(s.substring(p6 + 1, p7));
+ byte[] encodedEvent = java.util.Base64.getDecoder().decode(s.substring(p7 + 1));
+ return new Record(deliveryId, topic, partition, offset, clientId, attempt, nextAttemptAtMs, encodedEvent);
}
}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/state/DeliveryRecoveryTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/state/DeliveryRecoveryTest.java
index 3bb46d0d7..e3be9d568 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/state/DeliveryRecoveryTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/state/DeliveryRecoveryTest.java
@@ -69,19 +69,22 @@ class DeliveryRecoveryTest {
AtomicLong clock = new AtomicLong(10_000L);
ReliableDispatcher b = new ReliableDispatcher(1000L, 3, clock::get, offsets, dlq,
new UniMetrics(), 0.0d, store);
- int recovered = b.recover();
-
- // Each persisted delivery must be retired: offset advanced, store emptied
+ // Capture deliveries made during the simulated JVM so we can assert recovery did NOT
+ // re-deliver through the channel (the broker owns redelivery, issue #5291).
+ final int deliveredBeforeRecovery = channel.delivered.size();
+ final int recovered = b.recover();
assertEquals(3, recovered, "all 3 in-flight deliveries must be recovered");
- assertEquals(0, store.count(), "recovered entries are removed from the ledger");
- assertEquals(0, b.pendingCount(), "the fresh dispatcher must not re-track the recovered entries");
- assertEquals(100L, offsets.readOffset("topic-A", "client-X", 0));
- assertEquals(200L, offsets.readOffset("topic-B", "client-Y", 1));
// No channel redelivery happened during recovery (issue #5291 idempotency)
- assertEquals(0, channel.delivered.size(),
+ assertEquals(deliveredBeforeRecovery, channel.delivered.size(),
"recovery must NOT re-deliver through the channel (broker already redelivered or not, but "
+ "EventMesh is not the source of truth for the message anymore)");
+
+ // Each persisted delivery must be retired: offset advanced, store emptied
+ assertEquals(0, store.count(), "recovered entries are removed from the ledger");
+ assertEquals(0, b.pendingCount(), "the fresh dispatcher must not re-track the recovered entries");
+ assertEquals(101L, offsets.readOffset("topic-A", "client-X", 0));
+ assertEquals(200L, offsets.readOffset("topic-B", "client-Y", 1));
}
@Test
@@ -130,7 +133,6 @@ class DeliveryRecoveryTest {
}
private static class TestChannel implements PushChannel {
-
final List<String> delivered = new ArrayList<>();
@Override
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/state/DeliveryStateStoreTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/state/DeliveryStateStoreTest.java
index 5f6077243..1abd3d9d7 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/state/DeliveryStateStoreTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/state/DeliveryStateStoreTest.java
@@ -169,7 +169,7 @@ class DeliveryStateStoreTest {
private static Record newRecord(String deliveryId, String topic, int partition, long offset,
String clientId, int attempt, long nextAttemptAtMs) {
- EventMeshFrame event = EventMeshFrame.event(java.util.Map.of(), null);
+ EventMeshFrame event = EventMeshFrame.event(java.util.Map.of(), "payload".getBytes());
return new Record(deliveryId, topic, partition, offset, clientId, attempt, nextAttemptAtMs,
event.encode());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org