fix(offset): unify MQ physical offset and align pull cursor to ACK offset on restart
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 6896342..8935700 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 @@ 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 @@ private final MeshStoragePlugin storage; private final OffsetStore offsetStore; + private final PushOffsetStore pushOffsetStore; private final UniIngressService ingress; private final long pollIntervalMs; @@ -69,15 +72,31 @@ 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 @@ 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); @@ -101,6 +127,64 @@ } /** + * 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. */ public UniIngressService ingress() { @@ -179,7 +263,14 @@ } } - // 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 @@ 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 210b3ab..4adce8a 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
@@ -84,6 +84,12 @@ } @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(); flushDirty();
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 d8767c3..b0a6978 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 @@ 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 @@ 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 @@ 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 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 @@ /** * 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 @@ // 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 @@ * 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 @@ }); 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 @@ 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 - dist))); + 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, pushOff - ackOff))); } } } @@ -675,8 +736,11 @@ 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 caae7d2..59359a8 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
@@ -57,6 +57,18 @@ } @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 0000000..b81dd9d --- /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 48e013c..6feb01d 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
@@ -49,6 +49,15 @@ 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. */ void flush();
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 0000000..9f52321 --- /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 89a89e2..b5b0e68 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 @@ 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 e441a88..20c6d81 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 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 @@ 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 @@ 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 @@ 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 @@ 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 0000000..30a9683 --- /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/http/LegacyHttpServerIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/http/LegacyHttpServerIntegrationTest.java index 8eaa53b..a4e6427 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/http/LegacyHttpServerIntegrationTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/http/LegacyHttpServerIntegrationTest.java
@@ -23,6 +23,7 @@ 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.UniAdminService; import org.apache.eventmesh.runtime.boot.UniRuntime; import org.apache.eventmesh.runtime.delivery.CloudEventSerializer; @@ -45,11 +46,13 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; import io.cloudevents.CloudEvent; +import io.cloudevents.core.builder.CloudEventBuilder; /** * Real-HTTP integration: an old {@code EventMeshHttpClient} posts to {@code /eventmesh/publish} and @@ -97,7 +100,7 @@ assertEquals("hello-legacy", new String(webhook.posts.get(0).body, StandardCharsets.UTF_8)); // webhook returned 2xx → auto-ACK → offset advanced - assertTrue(runtime.ingress().getOffsetStore().readOffset("orders", "c1", -1) >= 1); + assertTrue(runtime.ingress().getOffsetStore().readOffset("orders", "c1", 0) >= 1); } private void boot() throws Exception { @@ -169,6 +172,7 @@ 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) { @@ -192,6 +196,12 @@ 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/ingress/UniIngressServiceTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/ingress/UniIngressServiceTest.java index bf3090a..3b1a34c 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 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 @@ 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 @@ 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 @@ 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 @@ 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 476ff26..812d608 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 @@ 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/BatchPublishIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/BatchPublishIntegrationTest.java index c6ff182..38fa380 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/BatchPublishIntegrationTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/BatchPublishIntegrationTest.java
@@ -23,6 +23,7 @@ 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.client.cloudevents.CloudEventsClient; import org.apache.eventmesh.runtime.admin.UniAdminService; import org.apache.eventmesh.runtime.http.UniHttpServer; @@ -39,12 +40,14 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import io.cloudevents.CloudEvent; +import io.cloudevents.core.builder.CloudEventBuilder; /** * In-process batch publish E2E: publish 10 events via {@code publishBatch}, verify all received via @@ -116,6 +119,7 @@ static final class InMemoryStorage implements MeshStoragePlugin { final ConcurrentHashMap<String, Queue<CloudEvent>> queues = new ConcurrentHashMap<>(); + final ConcurrentHashMap<String, AtomicLong> offsetSeq = new ConcurrentHashMap<>(); @Override public void init(java.util.Properties p) { @@ -139,6 +143,12 @@ 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/DlqIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/DlqIntegrationTest.java index c64ba3e..63230d3 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 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 @@ 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 @@ 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 @@ 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/RealBrokerIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/RealBrokerIntegrationTest.java index 4cada3a..25fbe6c 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/RealBrokerIntegrationTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/RealBrokerIntegrationTest.java
@@ -121,7 +121,7 @@ } // 4. The offset advanced only on ACK — the core at-least-once contract, now over real MQ. - long offset = runtime.ingress().getOffsetStore().readOffset(topic, clientId, -1); + long offset = runtime.ingress().getOffsetStore().readOffset(topic, clientId, 0); if (offset < 1) { throw new AssertionError("offset did not advance after ACK: " + offset); }
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/RequestReplyHttpIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/RequestReplyHttpIntegrationTest.java index e91621e..63549dc 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/RequestReplyHttpIntegrationTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/RequestReplyHttpIntegrationTest.java
@@ -23,6 +23,7 @@ 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.client.cloudevents.CloudEventsClient; import org.apache.eventmesh.runtime.admin.UniAdminService; import org.apache.eventmesh.runtime.http.UniHttpServer; @@ -38,12 +39,14 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import io.cloudevents.CloudEvent; +import io.cloudevents.core.builder.CloudEventBuilder; /** * In-process request-reply E2E over HTTP: a responder subscribes, a requester calls request(), @@ -123,6 +126,7 @@ static final class InMemoryStorage implements MeshStoragePlugin { final ConcurrentHashMap<String, Queue<CloudEvent>> queues = new ConcurrentHashMap<>(); + final ConcurrentHashMap<String, AtomicLong> offsetSeq = new ConcurrentHashMap<>(); @Override public void init(java.util.Properties p) { @@ -146,6 +150,12 @@ 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/WebSocketPushIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/WebSocketPushIntegrationTest.java index 7ad8391..825f8ab 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/WebSocketPushIntegrationTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/WebSocketPushIntegrationTest.java
@@ -22,6 +22,7 @@ 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.client.cloudevents.CloudEventsClient; import org.apache.eventmesh.runtime.admin.UniAdminService; import org.apache.eventmesh.runtime.http.UniHttpServer; @@ -39,6 +40,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -140,6 +142,7 @@ 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(java.util.Properties p) { @@ -163,6 +166,12 @@ 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 520390a..3c9ddaf 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 @@ 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 0000000..9679fb2 --- /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-runtime/src/test/java/org/apache/eventmesh/runtime/transport/http/LegacyHttpBridgeTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/http/LegacyHttpBridgeTest.java index d5605a9..769c26c 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/http/LegacyHttpBridgeTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/http/LegacyHttpBridgeTest.java
@@ -23,6 +23,7 @@ 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.boot.UniRuntime; import org.apache.eventmesh.runtime.delivery.HttpCaller; import org.apache.eventmesh.runtime.offset.InMemoryOffsetStore; @@ -39,6 +40,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; @@ -101,7 +103,7 @@ assertEquals("e-1", new String(httpCaller.posts.get(0).body, StandardCharsets.UTF_8)); // 4. the webhook returned 2xx → auto-ACK → offset advanced (at-least-once over legacy HTTP). - assertTrue(runtime.ingress().getOffsetStore().readOffset("orders", "c1", -1) >= 1, + assertTrue(runtime.ingress().getOffsetStore().readOffset("orders", "c1", 0) >= 1, "offset advanced after webhook delivery accepted"); } @@ -174,6 +176,7 @@ 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) { @@ -197,6 +200,12 @@ 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/transport/tcp/UniTcpServerTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServerTest.java index 880a25a..f906917 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServerTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServerTest.java
@@ -24,6 +24,7 @@ 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.common.protocol.tcp.Command; import org.apache.eventmesh.common.protocol.tcp.Header; import org.apache.eventmesh.common.protocol.tcp.Package; @@ -40,6 +41,7 @@ import java.util.Queue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicLong; import org.junit.jupiter.api.Test; @@ -140,7 +142,7 @@ clientAck.getHeader().putProperty(NettyTcpPushChannel.HEADER_DELIVERY_ID, deliveryId); client.writeInbound(clientAck); - assertEquals(1, ingress.getOffsetStore().readOffset("orders", "c1", -1), + assertEquals(1, ingress.getOffsetStore().readOffset("orders", "c1", 0), "offset advanced after the legacy TCP client ACKed the push"); assertEquals(1, ingress.getMetrics().getAckCount()); } @@ -153,6 +155,7 @@ private static final class InMemoryStorage implements MeshStoragePlugin { private final ConcurrentHashMap<String, Queue<CloudEvent>> queues = new ConcurrentHashMap<>(); + private final ConcurrentHashMap<String, AtomicLong> offsetSeq = new ConcurrentHashMap<>(); Queue<CloudEvent> queueOf(String topic) { return queues.computeIfAbsent(topic, k -> new ConcurrentLinkedQueue<>()); @@ -180,6 +183,12 @@ List<CloudEvent> out = new java.util.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-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 0844fc5..8dc25d7 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 @@ 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 0000000..9eb5ebd --- /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 f7d6853..6cc31cd 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 @@ 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.concurrent.ConcurrentHashMap; import io.cloudevents.CloudEvent; +import io.cloudevents.core.builder.CloudEventBuilder; import lombok.extern.slf4j.Slf4j; @@ -170,6 +172,11 @@ 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 @@ // 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 62edfb7..2327fd0 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.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.atomic.AtomicInteger; import io.cloudevents.CloudEvent; +import io.cloudevents.core.builder.CloudEventBuilder; import lombok.extern.slf4j.Slf4j; @@ -228,6 +230,11 @@ 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 @@ // 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 4fb9fa2..c5982ad 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.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.atomic.AtomicInteger; import io.cloudevents.CloudEvent; +import io.cloudevents.core.builder.CloudEventBuilder; import lombok.extern.slf4j.Slf4j; @@ -235,6 +237,11 @@ 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;