Skip to content

Commit 311dee5

Browse files
xwm1992xwm1992
committed
fix(offset): unify MQ physical offset and align pull cursor to ACK offset on restart
1 parent bd50ef7 commit 311dee5

20 files changed

Lines changed: 1100 additions & 23 deletions

File tree

‎eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/UniRuntime.java‎

Lines changed: 94 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,9 @@
1919

2020
import org.apache.eventmesh.api.storage.MeshStoragePlugin;
2121
import org.apache.eventmesh.runtime.ingress.UniIngressService;
22+
import org.apache.eventmesh.runtime.offset.InMemoryPushOffsetStore;
2223
import org.apache.eventmesh.runtime.offset.OffsetStore;
24+
import org.apache.eventmesh.runtime.offset.PushOffsetStore;
2325

2426
import java.util.Properties;
2527
import java.util.concurrent.Executors;
@@ -44,6 +46,7 @@ public class UniRuntime {
4446

4547
private final MeshStoragePlugin storage;
4648
private final OffsetStore offsetStore;
49+
private final PushOffsetStore pushOffsetStore;
4750
private final UniIngressService ingress;
4851

4952
private final long pollIntervalMs;
@@ -68,16 +71,32 @@ public UniRuntime withStorageConfig(Properties storageConfig) {
6871
}
6972

7073
public UniRuntime(MeshStoragePlugin storage, OffsetStore offsetStore,
74+
long pollIntervalMs, long tickIntervalMs, int maxBatchPerTopic, long pollTimeoutMs) {
75+
this(storage, offsetStore, new InMemoryPushOffsetStore(), pollIntervalMs, tickIntervalMs,
76+
maxBatchPerTopic, pollTimeoutMs);
77+
}
78+
79+
public UniRuntime(MeshStoragePlugin storage, OffsetStore offsetStore, PushOffsetStore pushOffsetStore,
7180
long pollIntervalMs, long tickIntervalMs, int maxBatchPerTopic, long pollTimeoutMs) {
7281
this.storage = storage;
7382
this.offsetStore = offsetStore;
74-
this.ingress = new UniIngressService(storage, offsetStore);
83+
this.pushOffsetStore = pushOffsetStore;
84+
this.ingress = new UniIngressService(storage, offsetStore, pushOffsetStore,
85+
new org.apache.eventmesh.runtime.subscription.SubscriptionManager(),
86+
new org.apache.eventmesh.runtime.push.PushService(),
87+
org.apache.eventmesh.runtime.delivery.ReliableDispatcher.DEFAULT_ACK_TIMEOUT_MS,
88+
org.apache.eventmesh.runtime.delivery.ReliableDispatcher.DEFAULT_MAX_ATTEMPTS,
89+
System::currentTimeMillis);
7590
this.pollIntervalMs = pollIntervalMs;
7691
this.tickIntervalMs = tickIntervalMs;
7792
this.maxBatchPerTopic = maxBatchPerTopic;
7893
this.pollTimeoutMs = pollTimeoutMs;
7994
}
8095

96+
public PushOffsetStore getPushOffsetStore() {
97+
return pushOffsetStore;
98+
}
99+
81100
/**
82101
* Start storage, offset store, and the pull/tick scheduler.
83102
*/
@@ -88,6 +107,13 @@ public void start() throws Exception {
88107
storage.init(storageConfig);
89108
storage.start();
90109

110+
// Restart recovery: align the storage's pull cursor to the ACK offset so that messages
111+
// pulled-but-not-ACKed before the restart are re-pulled (at-least-once). Without this,
112+
// the persisted pull offset (ahead of ACK offset after a crash) would skip the gap
113+
// messages — they are neither in the MQ's unconsumed range nor in the (lost) in-memory
114+
// pending deliveries.
115+
alignPullOffsetsToAck();
116+
91117
scheduler = Executors.newScheduledThreadPool(3, r -> {
92118
Thread t = new Thread(r, "eventmesh-uni");
93119
t.setDaemon(true);
@@ -100,6 +126,64 @@ public void start() throws Exception {
100126
log.info("uni runtime started (poll={}ms, tick={}ms)", pollIntervalMs, tickIntervalMs);
101127
}
102128

129+
/**
130+
* For every topic with persisted ACK offsets, rewind the storage plugin's pull cursor to the
131+
* minimum ACK offset across all clients for each partition. This ensures at-least-once delivery
132+
* after a restart: messages in the gap [ackOffset, pullOffset) are re-pulled and re-delivered.
133+
*
134+
* <p>Keyed by {@code clientId#partition} in {@link OffsetStore#readAllOffsets}, so the min ACK
135+
* offset per partition is computed across all clients that subscribed to that topic. Using min
136+
* (not max) guarantees the slowest client still receives its gap messages.</p>
137+
*
138+
* <p>Topics without any persisted ACK offset (first run / new topic) are skipped — the storage
139+
* plugin keeps its default cursor (beginning or latest per its own init logic).</p>
140+
*/
141+
private void alignPullOffsetsToAck() {
142+
// Discover all topics that have persisted ACK offsets via OffsetStore.readAllTopics().
143+
// This is the reliable recovery path: the store knows what it persisted across restarts.
144+
java.util.Set<String> topicsWithAckOffsets = offsetStore.readAllTopics();
145+
if (topicsWithAckOffsets.isEmpty()) {
146+
log.info("pull-offset alignment: no persisted ACK offsets (first run), skipping");
147+
return;
148+
}
149+
150+
int aligned = 0;
151+
for (String topic : topicsWithAckOffsets) {
152+
java.util.Map<String, Long> ackOffsets = offsetStore.readAllOffsets(topic);
153+
if (ackOffsets == null || ackOffsets.isEmpty()) {
154+
continue;
155+
}
156+
// Compute min ACK offset per partition across all clients.
157+
// Key format: clientId#partition → offset
158+
java.util.Map<Integer, Long> minAckByPartition = new java.util.HashMap<>();
159+
for (java.util.Map.Entry<String, Long> e : ackOffsets.entrySet()) {
160+
int sep = e.getKey().lastIndexOf('#');
161+
if (sep <= 0) {
162+
continue;
163+
}
164+
try {
165+
int partition = Integer.parseInt(e.getKey().substring(sep + 1));
166+
minAckByPartition.merge(partition, e.getValue(), Math::min);
167+
} catch (NumberFormatException ignored) {
168+
// key format mismatch — skip
169+
}
170+
}
171+
172+
for (java.util.Map.Entry<Integer, Long> e : minAckByPartition.entrySet()) {
173+
int partition = e.getKey();
174+
long ackOffset = e.getValue();
175+
if (ackOffset >= 0) {
176+
boolean rewound = storage.alignPullOffset(topic, partition, ackOffset);
177+
if (rewound) {
178+
aligned++;
179+
log.info("pull-offset alignment: {}#{} rewound to ACK offset {}", topic, partition, ackOffset);
180+
}
181+
}
182+
}
183+
}
184+
log.info("pull-offset alignment complete: {} partition(s) rewound across {} topic(s)", aligned, topicsWithAckOffsets.size());
185+
}
186+
103187
/**
104188
* The unified ingress — publish/subscribe/poll/ack/request-reply attach here.
105189
*/
@@ -179,7 +263,14 @@ public void shutdown(long gracefulMs) {
179263
}
180264
}
181265

182-
// 4. Flush + close offset store
266+
// 4. Clear push offset store (in-memory only, no persistence)
267+
try {
268+
pushOffsetStore.clear();
269+
} catch (Exception e) {
270+
log.warn("push offset store clear failed", e);
271+
}
272+
273+
// 5. Flush + close offset store
183274
try {
184275
offsetStore.flush();
185276
} catch (Exception e) {
@@ -191,7 +282,7 @@ public void shutdown(long gracefulMs) {
191282
log.warn("offset close failed", e);
192283
}
193284

194-
// 5. Close storage
285+
// 6. Close storage
195286
try {
196287
storage.shutdown();
197288
} catch (Exception e) {

‎eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/MetaBackedOffsetStore.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,12 @@ public Map<String, Long> readAllOffsets(String topic) {
8383
return local.readAllOffsets(topic);
8484
}
8585

86+
@Override
87+
public java.util.Set<String> readAllTopics() {
88+
// Delegate to local — it mirrors every write and is the crash-recovery source.
89+
return local.readAllTopics();
90+
}
91+
8692
@Override
8793
public void flush() {
8894
local.flush();

‎eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/UniIngressService.java‎

Lines changed: 80 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -20,12 +20,15 @@
2020
import org.apache.eventmesh.api.SendCallback;
2121
import org.apache.eventmesh.api.SendResult;
2222
import org.apache.eventmesh.api.storage.MeshStoragePlugin;
23+
import org.apache.eventmesh.api.storage.OffsetExtensions;
2324
import org.apache.eventmesh.runtime.delivery.DeadLetterSink;
2425
import org.apache.eventmesh.runtime.delivery.PushChannel;
2526
import org.apache.eventmesh.runtime.delivery.ReliableDispatcher;
2627
import org.apache.eventmesh.runtime.metrics.UniMetrics;
2728
import org.apache.eventmesh.runtime.metrics.UniTrace;
29+
import org.apache.eventmesh.runtime.offset.InMemoryPushOffsetStore;
2830
import org.apache.eventmesh.runtime.offset.OffsetStore;
31+
import org.apache.eventmesh.runtime.offset.PushOffsetStore;
2932
import org.apache.eventmesh.runtime.push.BufferedEvent;
3033
import org.apache.eventmesh.runtime.push.LongPollingChannel;
3134
import org.apache.eventmesh.runtime.push.PushService;
@@ -69,6 +72,7 @@ public class UniIngressService {
6972

7073
private final MeshStoragePlugin storage;
7174
private final OffsetStore offsetStore;
75+
private final PushOffsetStore pushOffsetStore;
7276
private final SubscriptionManager subscriptionManager;
7377
private final ReliableDispatcher dispatcher;
7478
private final PushService pushService;
@@ -84,7 +88,6 @@ public class UniIngressService {
8488
private final UniMetrics metrics;
8589

8690
private final ConcurrentHashMap<String, PushChannel> channels = new ConcurrentHashMap<>();
87-
private final ConcurrentHashMap<String, AtomicLong> topicOffsetSeq = new ConcurrentHashMap<>();
8891
private final ConcurrentHashMap<String, CompletableFuture<CloudEvent>> pendingRequests = new ConcurrentHashMap<>();
8992
private final AtomicLong requestSeq = new AtomicLong();
9093
private final ConcurrentHashMap<String, TokenBucketRateLimiter> topicLimiters = new ConcurrentHashMap<>();
@@ -99,19 +102,20 @@ public class UniIngressService {
99102
public static final String EXT_CORRELATION_ID = "emcorrelationid";
100103

101104
public UniIngressService(MeshStoragePlugin storage, OffsetStore offsetStore) {
102-
this(storage, offsetStore, new SubscriptionManager(), new PushService(),
105+
this(storage, offsetStore, new InMemoryPushOffsetStore(), new SubscriptionManager(), new PushService(),
103106
ReliableDispatcher.DEFAULT_ACK_TIMEOUT_MS, ReliableDispatcher.DEFAULT_MAX_ATTEMPTS,
104107
System::currentTimeMillis);
105108
}
106109

107110
/**
108111
* Test-friendly constructor with an injectable clock and retry parameters.
109112
*/
110-
public UniIngressService(MeshStoragePlugin storage, OffsetStore offsetStore,
113+
public UniIngressService(MeshStoragePlugin storage, OffsetStore offsetStore, PushOffsetStore pushOffsetStore,
111114
SubscriptionManager subscriptionManager, PushService pushService,
112115
long ackTimeoutMs, int maxAttempts, java.util.function.LongSupplier clock) {
113116
this.storage = storage;
114117
this.offsetStore = offsetStore;
118+
this.pushOffsetStore = pushOffsetStore;
115119
this.subscriptionManager = subscriptionManager;
116120
this.pushService = pushService;
117121
this.metrics = new UniMetrics();
@@ -299,9 +303,15 @@ private int pullAndDispatchPartition(String topic, int partition, int maxEvents,
299303
// Multi-instance: route via the cluster coordinator (local vs cross-instance forward).
300304
cluster.dispatch(topic, event);
301305
} else {
302-
long offset = nextOffset(topic);
306+
// Read MQ physical offset from CloudEvent extension (written by storage plugin on poll)
307+
long mqOffset = OffsetExtensions.readMqOffset(event);
308+
int mqPartition = OffsetExtensions.readMqPartition(event);
303309
for (Subscription target : subscriptionManager.targetsFor(topic, event)) {
304-
dispatcher.deliver(topic, partition, offset, event, target.getClientId(), channelFor(target.getClientId()));
310+
dispatcher.deliver(topic, mqPartition, mqOffset, event, target.getClientId(), channelFor(target.getClientId()));
311+
// Record push watermark for offset tracking
312+
if (mqOffset >= 0 && mqPartition >= 0) {
313+
pushOffsetStore.writePushOffset(topic, target.getClientId(), mqPartition, mqOffset);
314+
}
305315
}
306316
}
307317
UniTrace.end(dispatchSpan);
@@ -335,8 +345,14 @@ private boolean isExpired(CloudEvent event) {
335345
* same-instance targets here while forwarding remote ones.
336346
*/
337347
public boolean deliverLocal(String topic, String clientId, CloudEvent event) {
338-
long offset = nextOffset(topic);
339-
dispatcher.deliver(topic, -1, offset, event, clientId, channelFor(clientId));
348+
// Read MQ physical offset from CloudEvent extension (written by storage plugin on poll)
349+
long mqOffset = OffsetExtensions.readMqOffset(event);
350+
int mqPartition = OffsetExtensions.readMqPartition(event);
351+
dispatcher.deliver(topic, mqPartition, mqOffset, event, clientId, channelFor(clientId));
352+
// Record push watermark for offset tracking
353+
if (mqOffset >= 0 && mqPartition >= 0) {
354+
pushOffsetStore.writePushOffset(topic, clientId, mqPartition, mqOffset);
355+
}
340356
return true;
341357
}
342358

@@ -571,7 +587,7 @@ public void registerRuntimeGauges() {
571587
});
572588

573589
metrics.registerLabelledGauge("eventmesh_offset_lag",
574-
"MQ end offset - distributed offset (per topic/partition)",
590+
"MQ end offset - max ACK offset (per topic/partition) — total consumer lag",
575591
() -> {
576592
java.util.List<UniMetrics.LabelledLong> out = new java.util.ArrayList<>();
577593
if (partitionOwnership == null) {
@@ -582,27 +598,72 @@ public void registerRuntimeGauges() {
582598
if (owned == null) {
583599
continue;
584600
}
585-
// Max distributed offset per partition across all clients (key = clientId#partition).
586-
java.util.Map<Integer, Long> maxByPart = new java.util.HashMap<>();
601+
// Max ACK offset per partition across all clients (key = clientId#partition).
602+
java.util.Map<Integer, Long> maxAckByPart = new java.util.HashMap<>();
587603
for (java.util.Map.Entry<String, Long> e : offsetStore.readAllOffsets(t).entrySet()) {
588604
int sep = e.getKey().lastIndexOf('#');
589605
if (sep > 0) {
590606
try {
591607
int p = Integer.parseInt(e.getKey().substring(sep + 1));
592-
maxByPart.merge(p, e.getValue(), Math::max);
608+
maxAckByPart.merge(p, e.getValue(), Math::max);
593609
} catch (NumberFormatException expected) {
594610
}
595611
}
596612
}
597613
for (int p : owned) {
598614
long end = storage.endOffset(t, p);
599-
long dist = maxByPart.getOrDefault(p, -1L);
600-
if (end >= 0 && dist >= 0) {
615+
long ack = maxAckByPart.getOrDefault(p, -1L);
616+
if (end >= 0 && ack >= 0) {
617+
out.add(new UniMetrics.LabelledLong(
618+
io.opentelemetry.api.common.Attributes.of(
619+
io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t,
620+
io.opentelemetry.api.common.AttributeKey.longKey("partition"), (long) p),
621+
Math.max(0, end - ack)));
622+
}
623+
}
624+
}
625+
return out;
626+
});
627+
628+
metrics.registerLabelledGauge("eventmesh_push_ack_lag",
629+
"max push offset - max ACK offset (per topic/partition) — in-flight deliveries",
630+
() -> {
631+
java.util.List<UniMetrics.LabelledLong> out = new java.util.ArrayList<>();
632+
for (String t : subscriptionManager.activeTopics()) {
633+
// Max push offset per partition across all clients
634+
java.util.Map<Integer, Long> maxPushByPart = new java.util.HashMap<>();
635+
for (java.util.Map.Entry<String, Long> e : pushOffsetStore.readAllPushOffsets(t).entrySet()) {
636+
int sep = e.getKey().lastIndexOf('#');
637+
if (sep > 0) {
638+
try {
639+
int p = Integer.parseInt(e.getKey().substring(sep + 1));
640+
maxPushByPart.merge(p, e.getValue(), Math::max);
641+
} catch (NumberFormatException expected) {
642+
}
643+
}
644+
}
645+
// Max ACK offset per partition across all clients
646+
java.util.Map<Integer, Long> maxAckByPart = new java.util.HashMap<>();
647+
for (java.util.Map.Entry<String, Long> e : offsetStore.readAllOffsets(t).entrySet()) {
648+
int sep = e.getKey().lastIndexOf('#');
649+
if (sep > 0) {
650+
try {
651+
int p = Integer.parseInt(e.getKey().substring(sep + 1));
652+
maxAckByPart.merge(p, e.getValue(), Math::max);
653+
} catch (NumberFormatException expected) {
654+
}
655+
}
656+
}
657+
for (java.util.Map.Entry<Integer, Long> pushEntry : maxPushByPart.entrySet()) {
658+
int p = pushEntry.getKey();
659+
long pushOff = pushEntry.getValue();
660+
long ackOff = maxAckByPart.getOrDefault(p, -1L);
661+
if (pushOff >= 0 && ackOff >= 0) {
601662
out.add(new UniMetrics.LabelledLong(
602663
io.opentelemetry.api.common.Attributes.of(
603664
io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t,
604665
io.opentelemetry.api.common.AttributeKey.longKey("partition"), (long) p),
605-
Math.max(0, end - dist)));
666+
Math.max(0, pushOff - ackOff)));
606667
}
607668
}
608669
}
@@ -675,8 +736,11 @@ private PushChannel channelFor(String clientId) {
675736
return channels.computeIfAbsent(clientId, id -> new LongPollingChannel(pushService, id));
676737
}
677738

678-
private long nextOffset(String topic) {
679-
return topicOffsetSeq.computeIfAbsent(topic, k -> new AtomicLong()).incrementAndGet();
739+
/**
740+
* Expose PushOffsetStore for metrics and admin queries.
741+
*/
742+
public PushOffsetStore getPushOffsetStore() {
743+
return pushOffsetStore;
680744
}
681745

682746
private DeadLetterSink deadLetterSink() {

‎eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/InMemoryOffsetStore.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,18 @@ public Map<String, Long> readAllOffsets(String topic) {
5656
return result;
5757
}
5858

59+
@Override
60+
public java.util.Set<String> readAllTopics() {
61+
java.util.Set<String> topics = new java.util.HashSet<>();
62+
for (String key : table.keySet()) {
63+
int sep = key.indexOf('#');
64+
if (sep > 0) {
65+
topics.add(key.substring(0, sep));
66+
}
67+
}
68+
return topics;
69+
}
70+
5971
@Override
6072
public void flush() {
6173
// nothing buffered

0 commit comments

Comments
 (0)