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:55 UTC
(eventmesh) branch develop updated (36e8ff4fc -> 311dee574)
This is an automated email from the ASF dual-hosted git repository.
xwm1992 pushed a change to branch develop
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
omit 36e8ff4fc fix(offset): unify MQ physical offset and align pull cursor to ACK offset on restart
new 311dee574 fix(offset): unify MQ physical offset and align pull cursor to ACK offset on restart
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 (36e8ff4fc)
\
N -- N -- N refs/heads/develop (311dee574)
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:
.../eventmesh/runtime/ingress/UniIngressService.java | 1 +
.../apache/eventmesh/runtime/offset/PushOffsetStore.java | 10 ++++++----
.../eventmesh/runtime/admin/UniAdminServiceTest.java | 12 +++++++++++-
.../eventmesh/runtime/boot/PullOffsetAlignmentTest.java | 1 -
.../eventmesh/runtime/ingress/UniIngressServiceTest.java | 15 ++++++++++++---
.../runtime/it/AckTimeoutRedeliveryIntegrationTest.java | 1 +
.../apache/eventmesh/runtime/it/DlqIntegrationTest.java | 10 ++++++++++
.../apache/eventmesh/api/storage/OffsetExtensions.java | 6 +++---
8 files changed, 44 insertions(+), 12 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org
(eventmesh) 01/01: fix(offset): unify MQ physical offset and align pull cursor to ACK offset on restart
Posted by mi...@apache.org.
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