You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by mi...@apache.org on 2026/08/05 09:05:56 UTC
(eventmesh) 01/01: fix(offset): unify MQ physical offset and align pull cursor to ACK offset on restart
This is an automated email from the ASF dual-hosted git repository.
xwm1992 pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
commit 311dee574c1728fe93405dc9d7abcea581f72718
Author: xwm1992 <mi...@126.com>
AuthorDate: Wed Aug 5 17:05:27 2026 +0800
fix(offset): unify MQ physical offset and align pull cursor to ACK offset on restart
---
.../apache/eventmesh/runtime/boot/UniRuntime.java | 97 +++++++-
.../runtime/cluster/MetaBackedOffsetStore.java | 6 +
.../runtime/ingress/UniIngressService.java | 96 ++++++--
.../runtime/offset/InMemoryOffsetStore.java | 12 +
.../runtime/offset/InMemoryPushOffsetStore.java | 111 +++++++++
.../eventmesh/runtime/offset/OffsetStore.java | 9 +
.../eventmesh/runtime/offset/PushOffsetStore.java | 93 ++++++++
.../runtime/offset/RocksDBOffsetStore.java | 19 ++
.../runtime/admin/UniAdminServiceTest.java | 12 +-
.../runtime/boot/PullOffsetAlignmentTest.java | 264 +++++++++++++++++++++
.../runtime/ingress/UniIngressServiceTest.java | 15 +-
.../it/AckTimeoutRedeliveryIntegrationTest.java | 1 +
.../eventmesh/runtime/it/DlqIntegrationTest.java | 10 +
.../eventmesh/runtime/offset/OffsetStoreTest.java | 16 ++
.../runtime/offset/PushOffsetStoreTest.java | 113 +++++++++
.../eventmesh/api/storage/MeshStoragePlugin.java | 25 ++
.../eventmesh/api/storage/OffsetExtensions.java | 89 +++++++
.../kafka/storage/KafkaMeshStoragePlugin.java | 82 +++++++
.../storage/RocketMQRemotingStoragePlugin.java | 46 ++++
.../storage/RocketMQ5RemotingStoragePlugin.java | 7 +
20 files changed, 1100 insertions(+), 23 deletions(-)
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/UniRuntime.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/UniRuntime.java
index 689634286..893570031 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/UniRuntime.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/UniRuntime.java
@@ -19,7 +19,9 @@ package org.apache.eventmesh.runtime.boot;
import org.apache.eventmesh.api.storage.MeshStoragePlugin;
import org.apache.eventmesh.runtime.ingress.UniIngressService;
+import org.apache.eventmesh.runtime.offset.InMemoryPushOffsetStore;
import org.apache.eventmesh.runtime.offset.OffsetStore;
+import org.apache.eventmesh.runtime.offset.PushOffsetStore;
import java.util.Properties;
import java.util.concurrent.Executors;
@@ -44,6 +46,7 @@ public class UniRuntime {
private final MeshStoragePlugin storage;
private final OffsetStore offsetStore;
+ private final PushOffsetStore pushOffsetStore;
private final UniIngressService ingress;
private final long pollIntervalMs;
@@ -68,16 +71,32 @@ public class UniRuntime {
}
public UniRuntime(MeshStoragePlugin storage, OffsetStore offsetStore,
+ long pollIntervalMs, long tickIntervalMs, int maxBatchPerTopic, long pollTimeoutMs) {
+ this(storage, offsetStore, new InMemoryPushOffsetStore(), pollIntervalMs, tickIntervalMs,
+ maxBatchPerTopic, pollTimeoutMs);
+ }
+
+ public UniRuntime(MeshStoragePlugin storage, OffsetStore offsetStore, PushOffsetStore pushOffsetStore,
long pollIntervalMs, long tickIntervalMs, int maxBatchPerTopic, long pollTimeoutMs) {
this.storage = storage;
this.offsetStore = offsetStore;
- this.ingress = new UniIngressService(storage, offsetStore);
+ this.pushOffsetStore = pushOffsetStore;
+ this.ingress = new UniIngressService(storage, offsetStore, pushOffsetStore,
+ new org.apache.eventmesh.runtime.subscription.SubscriptionManager(),
+ new org.apache.eventmesh.runtime.push.PushService(),
+ org.apache.eventmesh.runtime.delivery.ReliableDispatcher.DEFAULT_ACK_TIMEOUT_MS,
+ org.apache.eventmesh.runtime.delivery.ReliableDispatcher.DEFAULT_MAX_ATTEMPTS,
+ System::currentTimeMillis);
this.pollIntervalMs = pollIntervalMs;
this.tickIntervalMs = tickIntervalMs;
this.maxBatchPerTopic = maxBatchPerTopic;
this.pollTimeoutMs = pollTimeoutMs;
}
+ public PushOffsetStore getPushOffsetStore() {
+ return pushOffsetStore;
+ }
+
/**
* Start storage, offset store, and the pull/tick scheduler.
*/
@@ -88,6 +107,13 @@ public class UniRuntime {
storage.init(storageConfig);
storage.start();
+ // Restart recovery: align the storage's pull cursor to the ACK offset so that messages
+ // pulled-but-not-ACKed before the restart are re-pulled (at-least-once). Without this,
+ // the persisted pull offset (ahead of ACK offset after a crash) would skip the gap
+ // messages — they are neither in the MQ's unconsumed range nor in the (lost) in-memory
+ // pending deliveries.
+ alignPullOffsetsToAck();
+
scheduler = Executors.newScheduledThreadPool(3, r -> {
Thread t = new Thread(r, "eventmesh-uni");
t.setDaemon(true);
@@ -100,6 +126,64 @@ public class UniRuntime {
log.info("uni runtime started (poll={}ms, tick={}ms)", pollIntervalMs, tickIntervalMs);
}
+ /**
+ * For every topic with persisted ACK offsets, rewind the storage plugin's pull cursor to the
+ * minimum ACK offset across all clients for each partition. This ensures at-least-once delivery
+ * after a restart: messages in the gap [ackOffset, pullOffset) are re-pulled and re-delivered.
+ *
+ * <p>Keyed by {@code clientId#partition} in {@link OffsetStore#readAllOffsets}, so the min ACK
+ * offset per partition is computed across all clients that subscribed to that topic. Using min
+ * (not max) guarantees the slowest client still receives its gap messages.</p>
+ *
+ * <p>Topics without any persisted ACK offset (first run / new topic) are skipped — the storage
+ * plugin keeps its default cursor (beginning or latest per its own init logic).</p>
+ */
+ private void alignPullOffsetsToAck() {
+ // Discover all topics that have persisted ACK offsets via OffsetStore.readAllTopics().
+ // This is the reliable recovery path: the store knows what it persisted across restarts.
+ java.util.Set<String> topicsWithAckOffsets = offsetStore.readAllTopics();
+ if (topicsWithAckOffsets.isEmpty()) {
+ log.info("pull-offset alignment: no persisted ACK offsets (first run), skipping");
+ return;
+ }
+
+ int aligned = 0;
+ for (String topic : topicsWithAckOffsets) {
+ java.util.Map<String, Long> ackOffsets = offsetStore.readAllOffsets(topic);
+ if (ackOffsets == null || ackOffsets.isEmpty()) {
+ continue;
+ }
+ // Compute min ACK offset per partition across all clients.
+ // Key format: clientId#partition → offset
+ java.util.Map<Integer, Long> minAckByPartition = new java.util.HashMap<>();
+ for (java.util.Map.Entry<String, Long> e : ackOffsets.entrySet()) {
+ int sep = e.getKey().lastIndexOf('#');
+ if (sep <= 0) {
+ continue;
+ }
+ try {
+ int partition = Integer.parseInt(e.getKey().substring(sep + 1));
+ minAckByPartition.merge(partition, e.getValue(), Math::min);
+ } catch (NumberFormatException ignored) {
+ // key format mismatch — skip
+ }
+ }
+
+ for (java.util.Map.Entry<Integer, Long> e : minAckByPartition.entrySet()) {
+ int partition = e.getKey();
+ long ackOffset = e.getValue();
+ if (ackOffset >= 0) {
+ boolean rewound = storage.alignPullOffset(topic, partition, ackOffset);
+ if (rewound) {
+ aligned++;
+ log.info("pull-offset alignment: {}#{} rewound to ACK offset {}", topic, partition, ackOffset);
+ }
+ }
+ }
+ }
+ log.info("pull-offset alignment complete: {} partition(s) rewound across {} topic(s)", aligned, topicsWithAckOffsets.size());
+ }
+
/**
* The unified ingress — publish/subscribe/poll/ack/request-reply attach here.
*/
@@ -179,7 +263,14 @@ public class UniRuntime {
}
}
- // 4. Flush + close offset store
+ // 4. Clear push offset store (in-memory only, no persistence)
+ try {
+ pushOffsetStore.clear();
+ } catch (Exception e) {
+ log.warn("push offset store clear failed", e);
+ }
+
+ // 5. Flush + close offset store
try {
offsetStore.flush();
} catch (Exception e) {
@@ -191,7 +282,7 @@ public class UniRuntime {
log.warn("offset close failed", e);
}
- // 5. Close storage
+ // 6. Close storage
try {
storage.shutdown();
} catch (Exception e) {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/MetaBackedOffsetStore.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/MetaBackedOffsetStore.java
index 210b3abe7..4adce8ac7 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/MetaBackedOffsetStore.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/MetaBackedOffsetStore.java
@@ -83,6 +83,12 @@ public class MetaBackedOffsetStore implements OffsetStore {
return local.readAllOffsets(topic);
}
+ @Override
+ public java.util.Set<String> readAllTopics() {
+ // Delegate to local — it mirrors every write and is the crash-recovery source.
+ return local.readAllTopics();
+ }
+
@Override
public void flush() {
local.flush();
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 d8767c3dc..b0a697871 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
@@ -20,12 +20,15 @@ package org.apache.eventmesh.runtime.ingress;
import org.apache.eventmesh.api.SendCallback;
import org.apache.eventmesh.api.SendResult;
import org.apache.eventmesh.api.storage.MeshStoragePlugin;
+import org.apache.eventmesh.api.storage.OffsetExtensions;
import org.apache.eventmesh.runtime.delivery.DeadLetterSink;
import org.apache.eventmesh.runtime.delivery.PushChannel;
import org.apache.eventmesh.runtime.delivery.ReliableDispatcher;
import org.apache.eventmesh.runtime.metrics.UniMetrics;
import org.apache.eventmesh.runtime.metrics.UniTrace;
+import org.apache.eventmesh.runtime.offset.InMemoryPushOffsetStore;
import org.apache.eventmesh.runtime.offset.OffsetStore;
+import org.apache.eventmesh.runtime.offset.PushOffsetStore;
import org.apache.eventmesh.runtime.push.BufferedEvent;
import org.apache.eventmesh.runtime.push.LongPollingChannel;
import org.apache.eventmesh.runtime.push.PushService;
@@ -69,6 +72,7 @@ public class UniIngressService {
private final MeshStoragePlugin storage;
private final OffsetStore offsetStore;
+ private final PushOffsetStore pushOffsetStore;
private final SubscriptionManager subscriptionManager;
private final ReliableDispatcher dispatcher;
private final PushService pushService;
@@ -84,7 +88,6 @@ public class UniIngressService {
private final UniMetrics metrics;
private final ConcurrentHashMap<String, PushChannel> channels = new ConcurrentHashMap<>();
- private final ConcurrentHashMap<String, AtomicLong> topicOffsetSeq = new ConcurrentHashMap<>();
private final ConcurrentHashMap<String, CompletableFuture<CloudEvent>> pendingRequests = new ConcurrentHashMap<>();
private final AtomicLong requestSeq = new AtomicLong();
private final ConcurrentHashMap<String, TokenBucketRateLimiter> topicLimiters = new ConcurrentHashMap<>();
@@ -99,7 +102,7 @@ public class UniIngressService {
public static final String EXT_CORRELATION_ID = "emcorrelationid";
public UniIngressService(MeshStoragePlugin storage, OffsetStore offsetStore) {
- this(storage, offsetStore, new SubscriptionManager(), new PushService(),
+ this(storage, offsetStore, new InMemoryPushOffsetStore(), new SubscriptionManager(), new PushService(),
ReliableDispatcher.DEFAULT_ACK_TIMEOUT_MS, ReliableDispatcher.DEFAULT_MAX_ATTEMPTS,
System::currentTimeMillis);
}
@@ -107,11 +110,12 @@ public class UniIngressService {
/**
* Test-friendly constructor with an injectable clock and retry parameters.
*/
- public UniIngressService(MeshStoragePlugin storage, OffsetStore offsetStore,
+ public UniIngressService(MeshStoragePlugin storage, OffsetStore offsetStore, PushOffsetStore pushOffsetStore,
SubscriptionManager subscriptionManager, PushService pushService,
long ackTimeoutMs, int maxAttempts, java.util.function.LongSupplier clock) {
this.storage = storage;
this.offsetStore = offsetStore;
+ this.pushOffsetStore = pushOffsetStore;
this.subscriptionManager = subscriptionManager;
this.pushService = pushService;
this.metrics = new UniMetrics();
@@ -299,9 +303,15 @@ public class UniIngressService {
// Multi-instance: route via the cluster coordinator (local vs cross-instance forward).
cluster.dispatch(topic, event);
} else {
- long offset = nextOffset(topic);
+ // Read MQ physical offset from CloudEvent extension (written by storage plugin on poll)
+ long mqOffset = OffsetExtensions.readMqOffset(event);
+ int mqPartition = OffsetExtensions.readMqPartition(event);
for (Subscription target : subscriptionManager.targetsFor(topic, event)) {
- dispatcher.deliver(topic, partition, offset, event, target.getClientId(), channelFor(target.getClientId()));
+ dispatcher.deliver(topic, mqPartition, mqOffset, event, target.getClientId(), channelFor(target.getClientId()));
+ // Record push watermark for offset tracking
+ if (mqOffset >= 0 && mqPartition >= 0) {
+ pushOffsetStore.writePushOffset(topic, target.getClientId(), mqPartition, mqOffset);
+ }
}
}
UniTrace.end(dispatchSpan);
@@ -335,8 +345,14 @@ public class UniIngressService {
* same-instance targets here while forwarding remote ones.
*/
public boolean deliverLocal(String topic, String clientId, CloudEvent event) {
- long offset = nextOffset(topic);
- dispatcher.deliver(topic, -1, offset, event, clientId, channelFor(clientId));
+ // Read MQ physical offset from CloudEvent extension (written by storage plugin on poll)
+ long mqOffset = OffsetExtensions.readMqOffset(event);
+ int mqPartition = OffsetExtensions.readMqPartition(event);
+ dispatcher.deliver(topic, mqPartition, mqOffset, event, clientId, channelFor(clientId));
+ // Record push watermark for offset tracking
+ if (mqOffset >= 0 && mqPartition >= 0) {
+ pushOffsetStore.writePushOffset(topic, clientId, mqPartition, mqOffset);
+ }
return true;
}
@@ -571,7 +587,7 @@ public class UniIngressService {
});
metrics.registerLabelledGauge("eventmesh_offset_lag",
- "MQ end offset - distributed offset (per topic/partition)",
+ "MQ end offset - max ACK offset (per topic/partition) — total consumer lag",
() -> {
java.util.List<UniMetrics.LabelledLong> out = new java.util.ArrayList<>();
if (partitionOwnership == null) {
@@ -582,27 +598,72 @@ public class UniIngressService {
if (owned == null) {
continue;
}
- // Max distributed offset per partition across all clients (key = clientId#partition).
- java.util.Map<Integer, Long> maxByPart = new java.util.HashMap<>();
+ // Max ACK offset per partition across all clients (key = clientId#partition).
+ java.util.Map<Integer, Long> maxAckByPart = new java.util.HashMap<>();
for (java.util.Map.Entry<String, Long> e : offsetStore.readAllOffsets(t).entrySet()) {
int sep = e.getKey().lastIndexOf('#');
if (sep > 0) {
try {
int p = Integer.parseInt(e.getKey().substring(sep + 1));
- maxByPart.merge(p, e.getValue(), Math::max);
+ maxAckByPart.merge(p, e.getValue(), Math::max);
} catch (NumberFormatException expected) {
}
}
}
for (int p : owned) {
long end = storage.endOffset(t, p);
- long dist = maxByPart.getOrDefault(p, -1L);
- if (end >= 0 && dist >= 0) {
+ long ack = maxAckByPart.getOrDefault(p, -1L);
+ if (end >= 0 && ack >= 0) {
+ out.add(new UniMetrics.LabelledLong(
+ io.opentelemetry.api.common.Attributes.of(
+ io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t,
+ io.opentelemetry.api.common.AttributeKey.longKey("partition"), (long) p),
+ Math.max(0, end - ack)));
+ }
+ }
+ }
+ return out;
+ });
+
+ metrics.registerLabelledGauge("eventmesh_push_ack_lag",
+ "max push offset - max ACK offset (per topic/partition) — in-flight deliveries",
+ () -> {
+ java.util.List<UniMetrics.LabelledLong> out = new java.util.ArrayList<>();
+ for (String t : subscriptionManager.activeTopics()) {
+ // Max push offset per partition across all clients
+ java.util.Map<Integer, Long> maxPushByPart = new java.util.HashMap<>();
+ for (java.util.Map.Entry<String, Long> e : pushOffsetStore.readAllPushOffsets(t).entrySet()) {
+ int sep = e.getKey().lastIndexOf('#');
+ if (sep > 0) {
+ try {
+ int p = Integer.parseInt(e.getKey().substring(sep + 1));
+ maxPushByPart.merge(p, e.getValue(), Math::max);
+ } catch (NumberFormatException expected) {
+ }
+ }
+ }
+ // Max ACK offset per partition across all clients
+ java.util.Map<Integer, Long> maxAckByPart = new java.util.HashMap<>();
+ for (java.util.Map.Entry<String, Long> e : offsetStore.readAllOffsets(t).entrySet()) {
+ int sep = e.getKey().lastIndexOf('#');
+ if (sep > 0) {
+ try {
+ int p = Integer.parseInt(e.getKey().substring(sep + 1));
+ maxAckByPart.merge(p, e.getValue(), Math::max);
+ } catch (NumberFormatException expected) {
+ }
+ }
+ }
+ for (java.util.Map.Entry<Integer, Long> pushEntry : maxPushByPart.entrySet()) {
+ int p = pushEntry.getKey();
+ long pushOff = pushEntry.getValue();
+ long ackOff = maxAckByPart.getOrDefault(p, -1L);
+ if (pushOff >= 0 && ackOff >= 0) {
out.add(new UniMetrics.LabelledLong(
io.opentelemetry.api.common.Attributes.of(
io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t,
io.opentelemetry.api.common.AttributeKey.longKey("partition"), (long) p),
- Math.max(0, end - dist)));
+ Math.max(0, pushOff - ackOff)));
}
}
}
@@ -675,8 +736,11 @@ public class UniIngressService {
return channels.computeIfAbsent(clientId, id -> new LongPollingChannel(pushService, id));
}
- private long nextOffset(String topic) {
- return topicOffsetSeq.computeIfAbsent(topic, k -> new AtomicLong()).incrementAndGet();
+ /**
+ * Expose PushOffsetStore for metrics and admin queries.
+ */
+ public PushOffsetStore getPushOffsetStore() {
+ return pushOffsetStore;
}
private DeadLetterSink deadLetterSink() {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/InMemoryOffsetStore.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/InMemoryOffsetStore.java
index caae7d2aa..59359a852 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/InMemoryOffsetStore.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/InMemoryOffsetStore.java
@@ -56,6 +56,18 @@ public class InMemoryOffsetStore implements OffsetStore {
return result;
}
+ @Override
+ public java.util.Set<String> readAllTopics() {
+ java.util.Set<String> topics = new java.util.HashSet<>();
+ for (String key : table.keySet()) {
+ int sep = key.indexOf('#');
+ if (sep > 0) {
+ topics.add(key.substring(0, sep));
+ }
+ }
+ return topics;
+ }
+
@Override
public void flush() {
// nothing buffered
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/InMemoryPushOffsetStore.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/InMemoryPushOffsetStore.java
new file mode 100644
index 000000000..b81dd9dc7
--- /dev/null
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/InMemoryPushOffsetStore.java
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.offset;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * In-memory {@link PushOffsetStore}. Thread-safe, not durable.
+ *
+ * <p>Per design decision #3, push offsets do not survive restart — the
+ * system relies on MQ pull offset (rewind) for recovery. This store is
+ * purely a runtime gauge for:
+ * <ul>
+ * <li>offset_lag = max(push_offset) - max(ack_offset) per topic/partition</li>
+ * <li>push watermark visibility in admin panel</li>
+ * </ul>
+ */
+public class InMemoryPushOffsetStore implements PushOffsetStore {
+
+ /**
+ * Primary index: {@code topic#clientId#partition → offset} (AtomicLong for concurrent writes).
+ */
+ private final ConcurrentHashMap<String, AtomicLong> table = new ConcurrentHashMap<>();
+
+ /**
+ * Secondary index for client-level cleanup: {@code clientId → set of composite keys}.
+ * Avoids full-table scan on {@link #removeClient}.
+ */
+ private final ConcurrentHashMap<String, java.util.Set<String>> clientKeys = new ConcurrentHashMap<>();
+
+ @Override
+ public void writePushOffset(String topic, String clientId, int partition, long offset) {
+ String key = PushOffsetStore.buildKey(topic, clientId, partition);
+ AtomicLong prev = table.computeIfAbsent(key, k -> new AtomicLong(Long.MIN_VALUE));
+ // Track max offset for this key (monotonic watermark)
+ long current = prev.get();
+ while (offset > current && !prev.compareAndSet(current, offset)) {
+ current = prev.get();
+ }
+ // Register key under client for fast removal
+ clientKeys.computeIfAbsent(clientId, k -> java.util.Collections.newSetFromMap(new ConcurrentHashMap<>()))
+ .add(key);
+ }
+
+ @Override
+ public long readPushOffset(String topic, String clientId, int partition) {
+ AtomicLong val = table.get(PushOffsetStore.buildKey(topic, clientId, partition));
+ return val == null ? -1L : val.get();
+ }
+
+ @Override
+ public long readMaxPushOffset(String topic, String clientId) {
+ long max = -1L;
+ String prefix = topic + "#" + clientId + "#";
+ for (Map.Entry<String, AtomicLong> e : table.entrySet()) {
+ if (e.getKey().startsWith(prefix)) {
+ long v = e.getValue().get();
+ if (v > max) {
+ max = v;
+ }
+ }
+ }
+ return max;
+ }
+
+ @Override
+ public Map<String, Long> readAllPushOffsets(String topic) {
+ String prefix = topic + "#";
+ Map<String, Long> result = new HashMap<>();
+ for (Map.Entry<String, AtomicLong> e : table.entrySet()) {
+ if (e.getKey().startsWith(prefix)) {
+ result.put(e.getKey().substring(prefix.length()), e.getValue().get());
+ }
+ }
+ return java.util.Collections.unmodifiableMap(result);
+ }
+
+ @Override
+ public void removeClient(String clientId) {
+ java.util.Set<String> keys = clientKeys.remove(clientId);
+ if (keys != null) {
+ for (String key : keys) {
+ table.remove(key);
+ }
+ }
+ }
+
+ @Override
+ public void clear() {
+ table.clear();
+ clientKeys.clear();
+ }
+}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/OffsetStore.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/OffsetStore.java
index 48e013c0a..6feb01db7 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/OffsetStore.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/OffsetStore.java
@@ -48,6 +48,15 @@ public interface OffsetStore {
*/
Map<String, Long> readAllOffsets(String topic);
+ /**
+ * All topics that have at least one persisted offset entry. Used by the restart recovery
+ * path ({@code UniRuntime.alignPullOffsetsToAck}) to discover which topics need pull-cursor
+ * rewind. Returns an empty set on first run (no offsets persisted yet).
+ */
+ default java.util.Set<String> readAllTopics() {
+ return java.util.Collections.emptySet();
+ }
+
/**
* Force any buffered writes to durable storage.
*/
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/PushOffsetStore.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/PushOffsetStore.java
new file mode 100644
index 000000000..9f523219d
--- /dev/null
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/PushOffsetStore.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.offset;
+
+import org.apache.eventmesh.api.storage.OffsetExtensions;
+
+import java.util.Map;
+
+/**
+ * Records the push watermark — the last MQ physical offset that has been
+ * successfully pushed to a subscriber (before ACK).
+ *
+ * <p>This is the "push offset" in the four-offset model:
+ * <pre>
+ * write offset (MQ physical offset at send time)
+ * pull offset (MQ physical offset at poll time)
+ * push offset (this store — offset handed to ReliableDispatcher.deliver)
+ * ACK offset (OffsetStore — offset confirmed by client ACK)
+ * </pre>
+ *
+ * <p>PushOffsetStore is <b>in-memory only</b> (per design decision #3): it
+ * serves as a real-time tracking gauge, not a recovery mechanism. On restart
+ * the watermarks are reset — the system recovers from MQ's pull offset
+ * (rewind / replay) instead of from the push watermark.</p>
+ *
+ * <p>Implementations must be thread-safe.</p>
+ */
+public interface PushOffsetStore {
+
+ /**
+ * Record that an event with the given MQ physical offset has been
+ * handed to the reliability layer for delivery to {@code clientId}.
+ *
+ * @param topic MQ topic
+ * @param clientId subscriber client id
+ * @param partition MQ partition / queue id
+ * @param offset MQ physical offset (from {@link OffsetExtensions})
+ */
+ void writePushOffset(String topic, String clientId, int partition, long offset);
+
+ /**
+ * @return the last pushed offset for (topic, clientId, partition),
+ * or {@code -1L} if never recorded
+ */
+ long readPushOffset(String topic, String clientId, int partition);
+
+ /**
+ * Read max pushed offset across all partitions for a given (topic, clientId).
+ * Used by the offset_lag gauge (offset_store vs push_store diff).
+ *
+ * @return max pushed offset, or {@code -1L} if never recorded
+ */
+ long readMaxPushOffset(String topic, String clientId);
+
+ /**
+ * All push offsets for a topic, keyed by {@code clientId#partition}.
+ *
+ * @return unmodifiable map
+ */
+ Map<String, Long> readAllPushOffsets(String topic);
+
+ /**
+ * Remove all push offset tracking for a client (e.g. on unsubscribe).
+ */
+ void removeClient(String clientId);
+
+ /**
+ * Clear all entries (used on shutdown / tests).
+ */
+ void clear();
+
+ /**
+ * The composite key used by implementations: {@code topic#clientId#partition}.
+ */
+ static String buildKey(String topic, String clientId, int partition) {
+ return topic + "#" + clientId + "#" + partition;
+ }
+}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/RocksDBOffsetStore.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/RocksDBOffsetStore.java
index 89a89e278..b5b0e6807 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/RocksDBOffsetStore.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/RocksDBOffsetStore.java
@@ -101,6 +101,25 @@ public class RocksDBOffsetStore implements OffsetStore {
return result;
}
+ /**
+ * Scan all keys in RocksDB and extract the unique topic prefixes (before the first {@code #}).
+ * Used by the restart recovery path to discover which topics have persisted ACK offsets.
+ */
+ @Override
+ public java.util.Set<String> readAllTopics() {
+ java.util.Set<String> topics = new java.util.HashSet<>();
+ try (RocksIterator it = db.newIterator()) {
+ for (it.seekToFirst(); it.isValid(); it.next()) {
+ String key = new String(it.key(), StandardCharsets.UTF_8);
+ int sep = key.indexOf('#');
+ if (sep > 0) {
+ topics.add(key.substring(0, sep));
+ }
+ }
+ }
+ return topics;
+ }
+
@Override
public void flush() {
try (FlushOptions flushOptions = new FlushOptions().setWaitForFlush(true)) {
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/admin/UniAdminServiceTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/admin/UniAdminServiceTest.java
index e441a88e8..20c6d8128 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/admin/UniAdminServiceTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/admin/UniAdminServiceTest.java
@@ -23,8 +23,10 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import org.apache.eventmesh.api.SendCallback;
import org.apache.eventmesh.api.SendResult;
import org.apache.eventmesh.api.storage.MeshStoragePlugin;
+import org.apache.eventmesh.api.storage.OffsetExtensions;
import org.apache.eventmesh.runtime.ingress.UniIngressService;
import org.apache.eventmesh.runtime.offset.InMemoryOffsetStore;
+import org.apache.eventmesh.runtime.offset.InMemoryPushOffsetStore;
import org.apache.eventmesh.runtime.push.PushService;
import org.apache.eventmesh.runtime.subscription.DistributionMode;
import org.apache.eventmesh.runtime.subscription.SubscriptionManager;
@@ -63,7 +65,7 @@ class UniAdminServiceTest {
svc.ack(polled.get(0).getDeliveryId());
assertEquals(0, admin.pendingDeliveries());
- assertTrue(admin.offsets("orders").containsKey("client-1#-1"), "offset recorded under clientId#partition");
+ assertTrue(admin.offsets("orders").containsKey("client-1#0"), "offset recorded under clientId#partition");
}
@Test
@@ -86,6 +88,7 @@ class UniAdminServiceTest {
AtomicLong clock = new AtomicLong(0L);
InMemoryStorage storage = new InMemoryStorage();
UniIngressService svc = new UniIngressService(storage, new InMemoryOffsetStore(),
+ new InMemoryPushOffsetStore(),
new SubscriptionManager(), new PushService(), 1_000L, 1, clock::get);
final UniAdminService admin = new UniAdminService(svc);
@@ -119,6 +122,7 @@ class UniAdminServiceTest {
private static final class InMemoryStorage implements MeshStoragePlugin {
private final ConcurrentHashMap<String, Queue<CloudEvent>> queues = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, AtomicLong> offsetSeq = new ConcurrentHashMap<>();
@Override
public void init(Properties properties) {
@@ -142,6 +146,12 @@ class UniAdminServiceTest {
List<CloudEvent> out = new ArrayList<>();
CloudEvent e;
while (out.size() < maxEvents && (e = q.poll()) != null) {
+ // Write MQ physical offset and partition to CloudEvent extensions for unified offset tracking
+ long offset = offsetSeq.computeIfAbsent(topic, k -> new AtomicLong()).incrementAndGet();
+ e = CloudEventBuilder.from(e)
+ .withExtension(OffsetExtensions.EM_MQ_OFFSET, offset)
+ .withExtension(OffsetExtensions.EM_MQ_PARTITION, 0)
+ .build();
out.add(e);
}
return out;
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/boot/PullOffsetAlignmentTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/boot/PullOffsetAlignmentTest.java
new file mode 100644
index 000000000..30a9683de
--- /dev/null
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/boot/PullOffsetAlignmentTest.java
@@ -0,0 +1,264 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.boot;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.storage.MeshStoragePlugin;
+import org.apache.eventmesh.runtime.offset.InMemoryOffsetStore;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+
+/**
+ * Verifies that {@link UniRuntime} rewinds the storage plugin's pull cursor to the ACK offset
+ * on startup, so that messages pulled-but-not-ACKed before a restart are re-pulled (at-least-once).
+ */
+class PullOffsetAlignmentTest {
+
+ private UniRuntime runtime;
+
+ @AfterEach
+ void tearDown() {
+ if (runtime != null) {
+ runtime.shutdown();
+ }
+ }
+
+ /**
+ * Scenario: OffsetStore has ACK offset 80 for (orders, client-1, partition 0).
+ * The storage plugin's pullOffsets (simulating file-recovered state) is at 100.
+ * On start(), UniRuntime should call alignPullOffset("orders", 0, 80) to rewind.
+ */
+ @Test
+ void alignsPullOffsetToAckOffsetOnStart() throws Exception {
+ // Pre-populate the ACK offset store with offset 80 (simulating RocksDB recovery)
+ InMemoryOffsetStore offsetStore = new InMemoryOffsetStore();
+ offsetStore.writeOffset("orders", "client-1", 0, 80L);
+
+ TrackingStorage storage = new TrackingStorage();
+ // Simulate the pull offset recovered from file = 100 (ahead of ACK 80)
+ storage.setPullOffset("orders", 0, 100L);
+
+ runtime = new UniRuntime(storage, offsetStore, 20L, 50L, 100, 50L);
+ runtime.start();
+
+ // Verify alignPullOffset was called with the correct ACK offset
+ List<AlignCall> calls = storage.getAlignCalls();
+ assertFalse(calls.isEmpty(), "alignPullOffset should have been called on start");
+ boolean found = false;
+ for (AlignCall c : calls) {
+ if ("orders".equals(c.topic) && c.partition == 0 && c.ackOffset == 80L) {
+ found = true;
+ break;
+ }
+ }
+ assertTrue(found, "expected alignPullOffset(orders, 0, 80) but got " + calls);
+
+ // Verify the pull offset was rewound from 100 to 80
+ assertEquals(80L, storage.getPullOffset("orders", 0),
+ "pull offset should be rewound to ACK offset 80");
+ }
+
+ /**
+ * Scenario: Multiple clients subscribed to the same topic with different ACK progress.
+ * client-1 ACKed to 80, client-2 ACKed to 50. The min ACK offset (50) should be used
+ * for rewind so the slowest client still receives its gap messages.
+ */
+ @Test
+ void usesMinAckOffsetAcrossClients() throws Exception {
+ InMemoryOffsetStore offsetStore = new InMemoryOffsetStore();
+ offsetStore.writeOffset("orders", "client-1", 0, 80L);
+ offsetStore.writeOffset("orders", "client-2", 0, 50L);
+
+ TrackingStorage storage = new TrackingStorage();
+ storage.setPullOffset("orders", 0, 100L);
+
+ runtime = new UniRuntime(storage, offsetStore, 20L, 50L, 100, 50L);
+ runtime.start();
+
+ // Should rewind to min(80, 50) = 50
+ assertEquals(50L, storage.getPullOffset("orders", 0),
+ "pull offset should be rewound to min ACK offset 50");
+ }
+
+ /**
+ * Scenario: No persisted ACK offsets (first run). alignPullOffset should NOT be called
+ * — the storage plugin keeps its default cursor.
+ */
+ @Test
+ void skipsAlignmentOnFirstRun() throws Exception {
+ InMemoryOffsetStore offsetStore = new InMemoryOffsetStore();
+ TrackingStorage storage = new TrackingStorage();
+
+ runtime = new UniRuntime(storage, offsetStore, 20L, 50L, 100, 50L);
+ runtime.start();
+
+ assertTrue(storage.getAlignCalls().isEmpty(),
+ "alignPullOffset should not be called when no ACK offsets exist");
+ }
+
+ /**
+ * Scenario: ACK offset (80) is ahead of pull offset (50) — no rewind needed.
+ * alignPullOffset should be called but the storage plugin returns false (no rewind).
+ */
+ @Test
+ void noRewindWhenAckAheadOfPull() throws Exception {
+ InMemoryOffsetStore offsetStore = new InMemoryOffsetStore();
+ offsetStore.writeOffset("orders", "client-1", 0, 80L);
+
+ TrackingStorage storage = new TrackingStorage();
+ storage.setPullOffset("orders", 0, 50L); // pull < ACK, no gap
+
+ runtime = new UniRuntime(storage, offsetStore, 20L, 50L, 100, 50L);
+ runtime.start();
+
+ // alignPullOffset is called, but storage should NOT rewind (50 < 80)
+ assertEquals(50L, storage.getPullOffset("orders", 0),
+ "pull offset should remain 50 when already behind ACK offset");
+ }
+
+ // ---- Test doubles ----
+
+ private static final class AlignCall {
+
+ final String topic;
+ final int partition;
+ final long ackOffset;
+
+ AlignCall(String topic, int partition, long ackOffset) {
+ this.topic = topic;
+ this.partition = partition;
+ this.ackOffset = ackOffset;
+ }
+
+ @Override
+ public String toString() {
+ return "AlignCall(" + topic + "#" + partition + " -> " + ackOffset + ")";
+ }
+ }
+
+ /**
+ * A minimal MeshStoragePlugin that tracks alignPullOffset calls and simulates a pull-offset
+ * cursor (like the file-recovered pullOffsets in Kafka/RocketMQ-4.x plugins).
+ */
+ private static final class TrackingStorage implements MeshStoragePlugin {
+
+ private final ConcurrentHashMap<String, ConcurrentHashMap<Integer, Long>> pullOffsets = new ConcurrentHashMap<>();
+ private final List<AlignCall> alignCalls = new ArrayList<>();
+
+ void setPullOffset(String topic, int partition, long offset) {
+ pullOffsets.computeIfAbsent(topic, k -> new ConcurrentHashMap<>()).put(partition, offset);
+ }
+
+ long getPullOffset(String topic, int partition) {
+ return pullOffsets.getOrDefault(topic, new ConcurrentHashMap<>()).getOrDefault(partition, -1L);
+ }
+
+ List<AlignCall> getAlignCalls() {
+ return alignCalls;
+ }
+
+ @Override
+ public boolean alignPullOffset(String topic, int partition, long ackOffset) {
+ alignCalls.add(new AlignCall(topic, partition, ackOffset));
+ if (ackOffset < 0) {
+ return false;
+ }
+ ConcurrentHashMap<Integer, Long> topicOffsets = pullOffsets.computeIfAbsent(topic, k -> new ConcurrentHashMap<>());
+ if (partition >= 0) {
+ Long current = topicOffsets.get(partition);
+ if (current != null && current <= ackOffset) {
+ return false; // no rewind needed
+ }
+ topicOffsets.put(partition, ackOffset);
+ return true;
+ }
+ return false;
+ }
+
+ // ---- rest is no-op / minimal ----
+
+ private final ConcurrentHashMap<String, ConcurrentLinkedQueue<CloudEvent>> queues = new ConcurrentHashMap<>();
+
+ @Override
+ public void init(Properties properties) {
+ }
+
+ @Override
+ public void send(String topic, CloudEvent event, SendCallback callback) {
+ queues.computeIfAbsent(topic, k -> new ConcurrentLinkedQueue<>()).offer(event);
+ SendResult r = new SendResult();
+ r.setMessageId(event.getId());
+ r.setTopic(topic);
+ callback.onSuccess(r);
+ }
+
+ @Override
+ public List<CloudEvent> poll(String topic, int partition, long startOffset, int maxEvents, long timeoutMs) {
+ ConcurrentLinkedQueue<CloudEvent> q = queues.get(topic);
+ if (q == null) {
+ return new ArrayList<>();
+ }
+ List<CloudEvent> out = new ArrayList<>();
+ CloudEvent e;
+ while (out.size() < maxEvents && (e = q.poll()) != null) {
+ out.add(e);
+ }
+ return out;
+ }
+
+ @Override
+ public void assignPartitions(String topic, List<Integer> partitions) {
+ }
+
+ @Override
+ public void commitOffset(String topic, int partition, long offset) {
+ }
+
+ @Override
+ public boolean isStarted() {
+ return true;
+ }
+
+ @Override
+ public boolean isClosed() {
+ return false;
+ }
+
+ @Override
+ public void start() {
+ }
+
+ @Override
+ public void shutdown() {
+ }
+ }
+}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/ingress/UniIngressServiceTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/ingress/UniIngressServiceTest.java
index bf3090af4..3b1a34ca4 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/ingress/UniIngressServiceTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/ingress/UniIngressServiceTest.java
@@ -26,7 +26,9 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import org.apache.eventmesh.api.SendCallback;
import org.apache.eventmesh.api.SendResult;
import org.apache.eventmesh.api.storage.MeshStoragePlugin;
+import org.apache.eventmesh.api.storage.OffsetExtensions;
import org.apache.eventmesh.runtime.offset.InMemoryOffsetStore;
+import org.apache.eventmesh.runtime.offset.InMemoryPushOffsetStore;
import org.apache.eventmesh.runtime.offset.OffsetStore;
import org.apache.eventmesh.runtime.push.BufferedEvent;
import org.apache.eventmesh.runtime.push.PushService;
@@ -67,7 +69,7 @@ class UniIngressServiceTest {
assertEquals("o-1", delivered.get(0).getEvent().getId());
assertTrue(svc.ack(delivered.get(0).getDeliveryId()));
- assertTrue(offsets.readOffset("orders", "client-1", -1) >= 1, "offset advances only on ACK");
+ assertTrue(offsets.readOffset("orders", "client-1", 0) >= 1, "offset advances only on ACK");
}
@Test
@@ -92,8 +94,8 @@ class UniIngressServiceTest {
AtomicLong clock = new AtomicLong(0L);
InMemoryStorage storage = new InMemoryStorage();
OffsetStore offsets = new InMemoryOffsetStore();
- UniIngressService svc = new UniIngressService(storage, offsets, new SubscriptionManager(),
- new PushService(), 10_000L, 3, clock::get);
+ UniIngressService svc = new UniIngressService(storage, offsets, new InMemoryPushOffsetStore(),
+ new SubscriptionManager(), new PushService(), 10_000L, 3, clock::get);
svc.subscribe("orders", "client-1", DistributionMode.BROADCAST, null);
svc.publish("orders", event("o-1")).get();
@@ -201,6 +203,7 @@ class UniIngressServiceTest {
private static final class InMemoryStorage implements MeshStoragePlugin {
private final ConcurrentHashMap<String, Queue<CloudEvent>> queues = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, AtomicLong> offsetSeq = new ConcurrentHashMap<>();
@Override
public void init(Properties properties) {
@@ -224,6 +227,12 @@ class UniIngressServiceTest {
List<CloudEvent> out = new ArrayList<>();
CloudEvent e;
while (out.size() < maxEvents && (e = q.poll()) != null) {
+ // Write MQ physical offset and partition to CloudEvent extensions for unified offset tracking
+ long offset = offsetSeq.computeIfAbsent(topic, k -> new AtomicLong()).incrementAndGet();
+ e = CloudEventBuilder.from(e)
+ .withExtension(OffsetExtensions.EM_MQ_OFFSET, offset)
+ .withExtension(OffsetExtensions.EM_MQ_PARTITION, 0)
+ .build();
out.add(e);
}
return out;
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/AckTimeoutRedeliveryIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/AckTimeoutRedeliveryIntegrationTest.java
index 476ff26ca..812d6087c 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/AckTimeoutRedeliveryIntegrationTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/AckTimeoutRedeliveryIntegrationTest.java
@@ -100,6 +100,7 @@ class AckTimeoutRedeliveryIntegrationTest {
storage.start();
// Test-friendly ingress: short ACK timeout + low maxAttempts so the test runs in seconds.
ingress = new UniIngressService(storage, new InMemoryOffsetStore(),
+ new org.apache.eventmesh.runtime.offset.InMemoryPushOffsetStore(),
new org.apache.eventmesh.runtime.subscription.SubscriptionManager(),
new org.apache.eventmesh.runtime.push.PushService(),
ACK_TIMEOUT_MS, MAX_ATTEMPTS, System::currentTimeMillis);
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/DlqIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/DlqIntegrationTest.java
index c64ba3e3b..63230d3cd 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/DlqIntegrationTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/DlqIntegrationTest.java
@@ -23,11 +23,13 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import org.apache.eventmesh.api.SendCallback;
import org.apache.eventmesh.api.SendResult;
import org.apache.eventmesh.api.storage.MeshStoragePlugin;
+import org.apache.eventmesh.api.storage.OffsetExtensions;
import org.apache.eventmesh.runtime.admin.UniAdminServer;
import org.apache.eventmesh.runtime.admin.UniAdminService;
import org.apache.eventmesh.runtime.http.UniHttpServer;
import org.apache.eventmesh.runtime.ingress.UniIngressService;
import org.apache.eventmesh.runtime.offset.InMemoryOffsetStore;
+import org.apache.eventmesh.runtime.offset.InMemoryPushOffsetStore;
import org.apache.eventmesh.runtime.push.BufferedEvent;
import org.apache.eventmesh.runtime.subscription.DistributionMode;
@@ -141,6 +143,7 @@ class DlqIntegrationTest {
storage = new InMemoryStorage();
// Test-friendly ingress: inject the clock so tick() advances without wall-clock waits.
ingress = new UniIngressService(storage, new InMemoryOffsetStore(),
+ new InMemoryPushOffsetStore(),
new org.apache.eventmesh.runtime.subscription.SubscriptionManager(),
new org.apache.eventmesh.runtime.push.PushService(),
1_000L, maxAttempts, clock::get);
@@ -156,6 +159,7 @@ class DlqIntegrationTest {
static final class InMemoryStorage implements MeshStoragePlugin {
private final ConcurrentHashMap<String, Queue<CloudEvent>> queues = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, AtomicLong> offsetSeq = new ConcurrentHashMap<>();
@Override
public void init(Properties p) {
@@ -180,6 +184,12 @@ class DlqIntegrationTest {
List<CloudEvent> out = new ArrayList<>();
CloudEvent e;
while (out.size() < maxEvents && (e = q.poll()) != null) {
+ // Write MQ physical offset and partition to CloudEvent extensions for unified offset tracking
+ long offset = offsetSeq.computeIfAbsent(topic, k -> new AtomicLong()).incrementAndGet();
+ e = CloudEventBuilder.from(e)
+ .withExtension(OffsetExtensions.EM_MQ_OFFSET, offset)
+ .withExtension(OffsetExtensions.EM_MQ_PARTITION, 0)
+ .build();
out.add(e);
}
return out;
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/offset/OffsetStoreTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/offset/OffsetStoreTest.java
index 520390a10..3c9ddaf3f 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/offset/OffsetStoreTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/offset/OffsetStoreTest.java
@@ -90,5 +90,21 @@ class OffsetStoreTest {
assertEquals(99L, all.get("worker-2#0"));
assertFalse(all.containsKey("worker-1#0".replace("0", "x")));
assertTrue(all.values().stream().allMatch(v -> v >= 0));
+
+ // readAllTopics returns the set of topics with persisted offsets.
+ java.util.Set<String> topics = store.readAllTopics();
+ assertTrue(topics.contains("orders"));
+ assertTrue(topics.contains("payments"));
+ assertEquals(2, topics.size());
+ }
+
+ /**
+ * readAllTopics on an empty store returns an empty set (first-run scenario).
+ */
+ @Test
+ void emptyStoreReturnsEmptyTopicSet() {
+ InMemoryOffsetStore store = new InMemoryOffsetStore();
+ assertTrue(store.readAllTopics().isEmpty());
+ store.close();
}
}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/offset/PushOffsetStoreTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/offset/PushOffsetStoreTest.java
new file mode 100644
index 000000000..9679fb217
--- /dev/null
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/offset/PushOffsetStoreTest.java
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.offset;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.jupiter.api.Test;
+
+class PushOffsetStoreTest {
+
+ @Test
+ void inMemoryContract() {
+ InMemoryPushOffsetStore store = new InMemoryPushOffsetStore();
+ contract(store);
+ store.clear();
+ }
+
+ /**
+ * A PushOffsetStore tracks the MAX offset (watermark) per key, not the last written value.
+ * Writing a lower offset after a higher one must NOT move the watermark back.
+ */
+ @Test
+ void watermarkNeverGoesBack() {
+ InMemoryPushOffsetStore store = new InMemoryPushOffsetStore();
+ store.writePushOffset("orders", "worker-1", 0, 100L);
+ assertEquals(100L, store.readPushOffset("orders", "worker-1", 0));
+
+ // Writing a lower offset must NOT move the watermark back
+ store.writePushOffset("orders", "worker-1", 0, 50L);
+ assertEquals(100L, store.readPushOffset("orders", "worker-1", 0));
+
+ // Writing a higher offset must advance the watermark
+ store.writePushOffset("orders", "worker-1", 0, 200L);
+ assertEquals(200L, store.readPushOffset("orders", "worker-1", 0));
+ }
+
+ /**
+ * readMaxPushOffset returns the max across all partitions for a (topic, clientId).
+ */
+ @Test
+ void maxPushOffsetAcrossPartitions() {
+ InMemoryPushOffsetStore store = new InMemoryPushOffsetStore();
+ store.writePushOffset("orders", "worker-1", 0, 10L);
+ store.writePushOffset("orders", "worker-1", 1, 30L);
+ store.writePushOffset("orders", "worker-1", 2, 20L);
+
+ assertEquals(30L, store.readMaxPushOffset("orders", "worker-1"));
+ assertEquals(-1L, store.readMaxPushOffset("orders", "worker-2"));
+ }
+
+ /**
+ * removeClient removes all entries for a client across all topics.
+ */
+ @Test
+ void removeClientCleansAllEntries() {
+ InMemoryPushOffsetStore store = new InMemoryPushOffsetStore();
+ store.writePushOffset("orders", "worker-1", 0, 10L);
+ store.writePushOffset("payments", "worker-1", 0, 20L);
+ store.writePushOffset("orders", "worker-2", 0, 30L);
+
+ store.removeClient("worker-1");
+
+ assertEquals(-1L, store.readPushOffset("orders", "worker-1", 0));
+ assertEquals(-1L, store.readPushOffset("payments", "worker-1", 0));
+ assertEquals(30L, store.readPushOffset("orders", "worker-2", 0)); // untouched
+ }
+
+ /**
+ * Shared contract every PushOffsetStore must satisfy.
+ */
+ private void contract(PushOffsetStore store) {
+ // Unknown offset reads as -1.
+ assertEquals(-1L, store.readPushOffset("orders", "worker-1", 0));
+
+ // Write then read is consistent.
+ store.writePushOffset("orders", "worker-1", 0, 10L);
+ store.writePushOffset("orders", "worker-1", 1, 11L);
+ store.writePushOffset("orders", "worker-2", 0, 99L);
+ // A different topic must not bleed across.
+ store.writePushOffset("payments", "worker-1", 0, 5L);
+
+ assertEquals(10L, store.readPushOffset("orders", "worker-1", 0));
+ assertEquals(11L, store.readPushOffset("orders", "worker-1", 1));
+ assertEquals(99L, store.readPushOffset("orders", "worker-2", 0));
+ assertEquals(-1L, store.readPushOffset("orders", "worker-3", 0));
+
+ // readAllPushOffsets returns only this topic's entries, keyed by clientId#partition.
+ java.util.Map<String, Long> all = store.readAllPushOffsets("orders");
+ assertEquals(3, all.size());
+ assertEquals(10L, all.get("worker-1#0"));
+ assertEquals(11L, all.get("worker-1#1"));
+ assertEquals(99L, all.get("worker-2#0"));
+ assertFalse(all.containsKey("worker-1#0".replace("0", "x")));
+ assertTrue(all.values().stream().allMatch(v -> v >= 0));
+ }
+}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-api/src/main/java/org/apache/eventmesh/api/storage/MeshStoragePlugin.java b/eventmesh-storage-plugin/eventmesh-storage-api/src/main/java/org/apache/eventmesh/api/storage/MeshStoragePlugin.java
index 0844fc534..8dc25d79b 100644
--- a/eventmesh-storage-plugin/eventmesh-storage-api/src/main/java/org/apache/eventmesh/api/storage/MeshStoragePlugin.java
+++ b/eventmesh-storage-plugin/eventmesh-storage-api/src/main/java/org/apache/eventmesh/api/storage/MeshStoragePlugin.java
@@ -123,4 +123,29 @@ public interface MeshStoragePlugin extends LifeCycle {
default long endOffset(String topic, int partition) {
return -1L;
}
+
+ /**
+ * Rewind the pull cursor for {@code (topic, partition)} to {@code ackOffset} so that messages
+ * already pulled but not yet ACKed by the client are re-pulled after a restart.
+ *
+ * <p>This is the recovery mechanism for the at-least-once contract on restart: the pull offset
+ * (persisted to a local file by Kafka/RocketMQ-4.x plugins) may be ahead of the ACK offset
+ * (persisted in RocksDB {@code OffsetStore}); without rewind, the gap messages are lost because
+ * they are neither in the MQ's unconsumed range nor in the in-memory {@code pending} deliveries
+ * (which is lost on restart).</p>
+ *
+ * <p>Implementations that manage their own pull cursor (Kafka {@code seek}, RocketMQ 4.x
+ * {@code pullOffsets}) MUST override this to rewind that cursor. Broker-managed backends
+ * (RocketMQ 5.x POP — broker re-delivers on invisible-timeout) can keep the default no-op.</p>
+ *
+ * @param topic EventMesh logical topic
+ * @param partition physical partition (-1 = all partitions of the topic)
+ * @param ackOffset the ACK offset to rewind to (from {@code OffsetStore}); {@code -1} means
+ * "no known ACK offset" → keep the existing pull cursor (new topic / first run)
+ * @return {@code true} if the cursor was rewound; {@code false} if the backend does not support
+ * rewind or {@code ackOffset} was not applicable
+ */
+ default boolean alignPullOffset(String topic, int partition, long ackOffset) {
+ return false;
+ }
}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-api/src/main/java/org/apache/eventmesh/api/storage/OffsetExtensions.java b/eventmesh-storage-plugin/eventmesh-storage-api/src/main/java/org/apache/eventmesh/api/storage/OffsetExtensions.java
new file mode 100644
index 000000000..9eb5ebde7
--- /dev/null
+++ b/eventmesh-storage-plugin/eventmesh-storage-api/src/main/java/org/apache/eventmesh/api/storage/OffsetExtensions.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.api.storage;
+
+/**
+ * CloudEvent extension attribute names for MQ physical offset propagation.
+ *
+ * <p>These extensions are written by storage plugins (Kafka / RocketMQ 4.x / 5.x)
+ * during {@code poll()}, so every CloudEvent flowing through EventMesh carries
+ * its origin MQ position. This eliminates EventMesh's self-generated logical
+ * sequence number and aligns all four offset categories to the MQ's physical
+ * offset:
+ * <pre>
+ * write offset (MQ physical offset at send time)
+ * pull offset (MQ physical offset at poll time)
+ * push offset (offset handed to ReliableDispatcher.deliver)
+ * ACK offset (offset confirmed by client ACK)
+ * </pre>
+ *
+ * <p>Extension names follow the CloudEvents spec: lower-case ASCII letters
+ * and digits only (no hyphens).</p>
+ */
+public final class OffsetExtensions {
+
+ /**
+ * MQ physical offset (long). Written by storage plugins on poll.
+ * Example: {@code event.getExtension(EM_MQ_OFFSET) → 123456L}
+ */
+ public static final String EM_MQ_OFFSET = "emmqoffset";
+
+ /**
+ * MQ partition / queue id (int). Written by storage plugins on poll.
+ * Example: {@code event.getExtension(EM_MQ_PARTITION) → 3}
+ */
+ public static final String EM_MQ_PARTITION = "emmqpartition";
+
+ private OffsetExtensions() {
+ // utility class
+ }
+
+ /**
+ * Read the MQ physical offset from a CloudEvent extension.
+ *
+ * @return the offset, or {@code -1L} if the extension is absent
+ */
+ public static long readMqOffset(io.cloudevents.CloudEvent event) {
+ Object v = event.getExtension(EM_MQ_OFFSET);
+ if (v == null) {
+ return -1L;
+ }
+ try {
+ return Long.parseLong(v.toString());
+ } catch (NumberFormatException e) {
+ return -1L;
+ }
+ }
+
+ /**
+ * Read the MQ partition / queue id from a CloudEvent extension.
+ *
+ * @return the partition, or {@code -1} if the extension is absent
+ */
+ public static int readMqPartition(io.cloudevents.CloudEvent event) {
+ Object v = event.getExtension(EM_MQ_PARTITION);
+ if (v == null) {
+ return -1;
+ }
+ try {
+ return Integer.parseInt(v.toString());
+ } catch (NumberFormatException e) {
+ return -1;
+ }
+ }
+}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/storage/KafkaMeshStoragePlugin.java b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/storage/KafkaMeshStoragePlugin.java
index f7d6853a1..6cc31cd1b 100644
--- a/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/storage/KafkaMeshStoragePlugin.java
+++ b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/storage/KafkaMeshStoragePlugin.java
@@ -20,6 +20,7 @@ package org.apache.eventmesh.storage.kafka.storage;
import org.apache.eventmesh.api.SendCallback;
import org.apache.eventmesh.api.SendResult;
import org.apache.eventmesh.api.storage.MeshStoragePlugin;
+import org.apache.eventmesh.api.storage.OffsetExtensions;
import java.util.ArrayList;
import java.util.Collections;
@@ -29,6 +30,7 @@ import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
import lombok.extern.slf4j.Slf4j;
@@ -170,6 +172,11 @@ public class KafkaMeshStoragePlugin implements MeshStoragePlugin {
pullOffsets.computeIfAbsent(topic, k -> new ConcurrentHashMap<>()).put(record.partition(), record.offset());
CloudEvent event = deserialize(record.value());
if (event != null) {
+ // Write MQ physical offset and partition to CloudEvent extensions for unified offset tracking
+ event = CloudEventBuilder.from(event)
+ .withExtension(OffsetExtensions.EM_MQ_OFFSET, record.offset())
+ .withExtension(OffsetExtensions.EM_MQ_PARTITION, record.partition())
+ .build();
events.add(event);
}
}
@@ -202,6 +209,81 @@ public class KafkaMeshStoragePlugin implements MeshStoragePlugin {
// This is intentionally a no-op.
}
+ /**
+ * Rewind the Kafka consumer's pull cursor for {@code (topic, partition)} to {@code ackOffset}.
+ *
+ * <p>On restart, the persisted {@code pullOffsets} file may be ahead of the ACK offset stored
+ * in RocksDB (messages pulled but not yet ACKed by the client). Without rewind, those messages
+ * are lost: they are no longer in Kafka's unconsumed range (seek would skip them) and the
+ * in-memory {@code pending} deliveries were lost on restart.</p>
+ *
+ * <p>This method must be called <em>before</em> the first {@link #poll} for the topic, so the
+ * lazy assignment + seek in {@code poll()} picks up the rewound offset. It overwrites both the
+ * in-memory {@code pullOffsets} (which was loaded from the persisted file in {@link #init})
+ * and, if the consumer is already assigned, issues a direct {@code seek}.</p>
+ */
+ @Override
+ public synchronized boolean alignPullOffset(String topic, int partition, long ackOffset) {
+ if (!consumerReady || ackOffset < 0) {
+ return false;
+ }
+ ConcurrentHashMap<Integer, Long> topicOffsets = pullOffsets.computeIfAbsent(topic, k -> new ConcurrentHashMap<>());
+ if (partition >= 0) {
+ // Single partition rewind
+ Long current = topicOffsets.get(partition);
+ if (current != null && current <= ackOffset) {
+ // Pull cursor is at or behind ACK offset — no rewind needed (no gap messages)
+ return false;
+ }
+ topicOffsets.put(partition, ackOffset);
+ seekIfAssigned(topic, partition, ackOffset);
+ log.info("aligned pull offset for {}#{}: {} -> {}", topic, partition, current, ackOffset);
+ } else {
+ // partition -1: rewind all partitions of this topic
+ java.util.Set<org.apache.kafka.common.TopicPartition> tps = assignedTopics.get(topic);
+ if (tps == null) {
+ // Not yet assigned — poll()'s lazy init will seek to pullOffsets, so just overwrite
+ for (java.util.Map.Entry<Integer, Long> e : topicOffsets.entrySet()) {
+ if (e.getValue() > ackOffset) {
+ e.setValue(ackOffset);
+ }
+ }
+ log.info("aligned pull offset for {} (all partitions, pre-assign) to <= {}", topic, ackOffset);
+ return !topicOffsets.isEmpty();
+ }
+ boolean anyRewound = false;
+ for (org.apache.kafka.common.TopicPartition tp : tps) {
+ Long current = topicOffsets.get(tp.partition());
+ if (current != null && current > ackOffset) {
+ topicOffsets.put(tp.partition(), ackOffset);
+ consumer.seek(tp, ackOffset);
+ log.info("aligned pull offset for {}#{}: {} -> {}", topic, tp.partition(), current, ackOffset);
+ anyRewound = true;
+ }
+ }
+ return anyRewound;
+ }
+ return true;
+ }
+
+ /**
+ * If the consumer is already assigned to {@code (topic, partition)}, issue a direct
+ * {@code consumer.seek}; otherwise the next {@link #poll} will pick up the updated
+ * {@code pullOffsets} entry during its lazy assignment.
+ */
+ private void seekIfAssigned(String topic, int partition, long offset) {
+ java.util.Set<org.apache.kafka.common.TopicPartition> tps = assignedTopics.get(topic);
+ if (tps == null) {
+ return;
+ }
+ for (org.apache.kafka.common.TopicPartition tp : tps) {
+ if (tp.partition() == partition) {
+ consumer.seek(tp, offset);
+ return;
+ }
+ }
+ }
+
@Override
public synchronized int partitionCount(String topic) {
if (!consumerReady) {
diff --git a/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/storage/RocketMQRemotingStoragePlugin.java b/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/storage/RocketMQRemotingStoragePlugin.java
index 62edfb746..2327fd073 100644
--- a/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/storage/RocketMQRemotingStoragePlugin.java
+++ b/eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/storage/RocketMQRemotingStoragePlugin.java
@@ -22,6 +22,7 @@ import org.apache.eventmesh.api.SendResult;
import org.apache.eventmesh.api.exception.OnExceptionContext;
import org.apache.eventmesh.api.exception.StorageRuntimeException;
import org.apache.eventmesh.api.storage.MeshStoragePlugin;
+import org.apache.eventmesh.api.storage.OffsetExtensions;
import java.util.ArrayList;
import java.util.Collections;
@@ -34,6 +35,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
import lombok.extern.slf4j.Slf4j;
@@ -228,6 +230,11 @@ public class RocketMQRemotingStoragePlugin implements MeshStoragePlugin {
for (org.apache.rocketmq.common.message.MessageExt msg : msgs) {
CloudEvent event = deserialize(msg.getBody());
if (event != null) {
+ // Write MQ physical offset and partition to CloudEvent extensions for unified offset tracking
+ event = CloudEventBuilder.from(event)
+ .withExtension(OffsetExtensions.EM_MQ_OFFSET, msg.getQueueOffset())
+ .withExtension(OffsetExtensions.EM_MQ_PARTITION, (long) msg.getQueueId())
+ .build();
events.add(event);
}
}
@@ -261,6 +268,45 @@ public class RocketMQRemotingStoragePlugin implements MeshStoragePlugin {
// Self-managed via pullOffsets + persisted to file.
}
+ /**
+ * Rewind the RocketMQ 4.x pull cursor for {@code (topic, partition)} to {@code ackOffset}.
+ *
+ * <p>On restart, the persisted {@code pullOffsets} file (nextBeginOffset per topic#queueId) may
+ * be ahead of the ACK offset stored in RocksDB. Without rewind, the gap messages (pulled but
+ * not ACKed) are lost — the next {@link #poll} resumes from the persisted nextBeginOffset,
+ * skipping them.</p>
+ *
+ * <p>This overwrites the in-memory {@code pullOffsets} so the next {@link #poll} uses
+ * {@code ackOffset} as the {@code queueOffset} in the PULL_MESSAGE request.</p>
+ */
+ @Override
+ public boolean alignPullOffset(String topic, int partition, long ackOffset) {
+ if (remotingClient == null || ackOffset < 0) {
+ return false;
+ }
+ ConcurrentHashMap<Integer, Long> topicOffsets = pullOffsets.computeIfAbsent(topic, k -> new ConcurrentHashMap<>());
+ if (partition >= 0) {
+ Long current = topicOffsets.get(partition);
+ if (current != null && current <= ackOffset) {
+ return false;
+ }
+ topicOffsets.put(partition, ackOffset);
+ log.info("aligned pull offset for {}#{}: {} -> {}", topic, partition, current, ackOffset);
+ } else {
+ // partition -1: rewind all queues of this topic
+ boolean anyRewound = false;
+ for (Map.Entry<Integer, Long> e : topicOffsets.entrySet()) {
+ if (e.getValue() > ackOffset) {
+ log.info("aligned pull offset for {}#{}: {} -> {}", topic, e.getKey(), e.getValue(), ackOffset);
+ e.setValue(ackOffset);
+ anyRewound = true;
+ }
+ }
+ return anyRewound;
+ }
+ return true;
+ }
+
@Override
public boolean isStarted() {
return remotingClient != null;
diff --git a/eventmesh-storage-plugin/eventmesh-storage-rocketmq5/src/main/java/org/apache/eventmesh/storage/rocketmq5/storage/RocketMQ5RemotingStoragePlugin.java b/eventmesh-storage-plugin/eventmesh-storage-rocketmq5/src/main/java/org/apache/eventmesh/storage/rocketmq5/storage/RocketMQ5RemotingStoragePlugin.java
index 4fb9fa244..c5982adc2 100644
--- a/eventmesh-storage-plugin/eventmesh-storage-rocketmq5/src/main/java/org/apache/eventmesh/storage/rocketmq5/storage/RocketMQ5RemotingStoragePlugin.java
+++ b/eventmesh-storage-plugin/eventmesh-storage-rocketmq5/src/main/java/org/apache/eventmesh/storage/rocketmq5/storage/RocketMQ5RemotingStoragePlugin.java
@@ -23,6 +23,7 @@ import org.apache.eventmesh.api.exception.OnExceptionContext;
import org.apache.eventmesh.api.exception.StorageRuntimeException;
import org.apache.eventmesh.api.storage.LiteTopicCapable;
import org.apache.eventmesh.api.storage.MeshStoragePlugin;
+import org.apache.eventmesh.api.storage.OffsetExtensions;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
@@ -56,6 +57,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
import lombok.extern.slf4j.Slf4j;
@@ -235,6 +237,11 @@ public class RocketMQ5RemotingStoragePlugin implements MeshStoragePlugin, LiteTo
ackNormal(brokerAddr, msg);
CloudEvent event = deserialize(msg.getBody());
if (event != null) {
+ // Write MQ physical offset and partition to CloudEvent extensions for unified offset tracking
+ event = CloudEventBuilder.from(event)
+ .withExtension(OffsetExtensions.EM_MQ_OFFSET, msg.getQueueOffset())
+ .withExtension(OffsetExtensions.EM_MQ_PARTITION, (long) msg.getQueueId())
+ .build();
events.add(event);
if (events.size() >= maxEvents) {
break;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org