fix(cluster): unify delivery topology with Meta CAS + fencing (#5293 #5288) (#5308)
Sticky-only delivery topology (removes cross-instance forwarding):
- Delete HttpForwarder and the /internal/forward + /internal/reply-forward
endpoints in UniHttpServer; reply() on a non-owner instance now 404s
(SDK pins to one instance via instanceUrl, no forwarding needed)
- Remove DistributionMode.LOAD_BALANCE_STICKY; LOAD_BALANCE absorbs the
sticky semantics (partitionkey hash routes to one subscriber)
Atomic Meta CAS fencing (replaces read-then-write gen overwrite):
- MetaStore.tryAcquire(key, expectedOldValue, newValue): atomic CAS
(InMemory: ConcurrentHashMap putIfAbsent/replace; Nacos 2.x:
publishConfigCas with casMd5 = MD5(expected content))
- FencingToken "<bootEpoch>:<counter>" per JVM, monotonic and
restart-safe; assignment record "/em/assignments/<topic#partition>"
= "<token>|<ownerInstanceId>"
- acquireOrFence: Case 1 unclaimed (or released tombstone) -> CAS claim;
Case 2 still ours -> sync token; Case 3 another owner -> CAS takeover
when TTL-evicted or token strictly higher, else fenced
- releaseStale: partitions that left our assigner share are CAS'd to a
"" tombstone so the new rightful owner can claim them (prevents
stranding after membership churn); tombstone is CAS-equivalent to
absent on both MetaStore backends
Heartbeat scheduling fix (#5288):
- ClusterMembership.heartbeat() was never scheduled; enableCluster now
runs it every 5s and releases it on shutdown together with the
partition lease
Tests:
- FencingTokenTest, InMemoryMetaStoreTest (incl. concurrent CAS),
PartitionFencingTest (first claim / race / restart fencing / stale CAS)
- ClusterDeliveryFaultTest: in-process 3-4 instance fault injection
(steady-state split, crash takeover, scale-out churn, Meta partition
split-brain guard, healed partition reclaim) driven by a mutable
clock - no sleeps, fully deterministic
- Remove ClusterForwardIntegrationTest / NacosClusterForwardIntegrationTest
(they covered the deleted forwarding path)
diff --git a/docs/eventmesh-uni-architecture-redesign.md b/docs/eventmesh-uni-architecture-redesign.md
index 02bbae0..fec8927 100644
--- a/docs/eventmesh-uni-architecture-redesign.md
+++ b/docs/eventmesh-uni-architecture-redesign.md
@@ -2872,6 +2872,79 @@
> **降级哲学**：Meta 挂时"尽力而为 + 不丢数据"，牺牲部分一致性（新订阅、故障接管）换取可用性；恢复后渐进对齐，幂等兜底收敛。这是 §15 "可降级部署"原则在多实例协调上的落实。
+#### 13.2.10 统一投递拓扑：sticky 模型 + Meta CAS fencing（#5293 实装）
+
+> **🔎 实现状态（v1.12 / 2026-08-19）**：✅ 已实现。删除跨实例转发路径与 `LOAD_BALANCE_STICKY` 模式；分区所有权改为 Meta CAS + `FencingToken`（替代 gen 数字）；心跳调度补齐（#5288）。含故障注入测试 `ClusterDeliveryFaultTest`（in-process 3-4 实例：稳态分配 / 宕机接管 / 扩容防搁浅 / Meta 分区脑裂防护 / 分区愈合）。
+
+**① 投递拓扑统一为 sticky（删除转发路径）**
+
+此前架构同时存在两条下发路径：分区 owner 实例拉取后**本地下发**，或**跨实例转发**给订阅者所在实例（`HttpForwarder` + `/internal/forward` 端点）。双路径导致：订阅漂移时序复杂、转发故障域大、`LOAD_BALANCE_STICKY` 语义与转发耦合。
+
+**决定**：只保留 sticky 单路径——
+
+```
+删除：
+ · HttpForwarder（整个类）+ UniHttpServer 的 /internal/forward、/internal/reply-forward 端点
+ · EventMeshApplication 中转发相关 wiring
+ · DistributionMode.LOAD_BALANCE_STICKY 枚举值（破坏性变更，模式合并）
+
+模型：
+ · 每实例只拉取自己 OWN 的分区（PartitionOwnership），本地下发给本实例订阅者
+ · 订阅者通过 /events/subscribe 返回的 instanceUrl 固定（pin）到一个实例
+ → SDK 的 poll/ack 永远落在同一实例，无跨实例转发需求
+ · LOAD_BALANCE 吸收原 LOAD_BALANCE_STICKY 行为：事件带 partitionkey 属性时
+ hash(partitionkey) 稳定路由到一个订阅者（保序），否则 round-robin
+```
+
+**② Meta CAS fencing：`tryAcquire` + `FencingToken`（替代 gen）**
+
+§13.2.8 ④ 原设计用自增 gen 数字做 fencing，但旧实现的读写是 read-then-write（非原子）：两实例同时读到 `null` 会双双 `put`，后写者静默获胜——fencing 失效。
+
+**实装**：
+
+```
+MetaStore 新增原子 CAS 接口：
+ boolean tryAcquire(String key, String expectedOldValue, String newValue)
+ · expectedOldValue == null → 键必须不存在（首claim）
+ · 实现：Nacos 2.x publishConfigCas(dataId, group, content, casMd5)
+ casMd5 = MD5(expectedOldValue == null ? "" : expectedOldValue)
+ InMemoryMetaStore → ConcurrentHashMap.replace(key, old, new) / putIfAbsent
+
+FencingToken（每 JVM 一个，单调递增）：
+ · 格式 "<bootEpoch>:<counter>"，bootEpoch = 启动毫秒时间戳，counter 原子自增
+ · 排序：先比 bootEpoch（旧 JVM 永远输），同 epoch 比 counter
+ · 存活于 Meta：/em/assignments/<topic#partition> = "<token>|<ownerInstanceId>"
+
+acquireOrFence 协议（PartitionOwnership）：
+ Case 1 键不存在（或为释放墓碑 ""）→ tryAcquire(currentValue → myToken|self)；CAS 失败 = 输了竞争，下轮再读
+ Case 2 owner 是自己 → 同步本地 token，继续持有
+ Case 3 owner 是别人 → 接管条件（满足其一即 tryAcquire(currentValue → myToken|self)）：
+ · owner 已被 TTL 驱逐（不在 live set）→ 强制接管
+ （仍轮询的僵尸实例必然已心跳失败、leaseValid=false 停止轮询，强制接管安全）
+ · myToken > metaToken（CAS fencing）
+ 否则自己被 fence，停止 poll 该分区
+
+释放路径 releaseStale（成员变更防搁浅）：
+ · 分区离开本实例的 assigner 份额（扩缩容改变取模映射）而 Meta 记录仍指向自己
+ → CAS 到释放墓碑 ""（仅当记录仍指向自己，不会破坏并发接管）
+ · 新 rightful owner 下轮以 Case 1 认领；否则旧 owner 的较高 token 会把新 owner
+ 永久 fence（分区搁浅，无人拉取）
+ · 墓碑 "" 与键不存在在 CAS 语义上等价（Nacos casMd5 = MD5("") 双向兼容）
+```
+
+**③ 心跳调度补齐（#5288 修复）**
+
+`ClusterMembership.heartbeat()` 此前从未被调度执行（`EventMeshApplication` 没有任何调用点），导致 `/session/recommend` 永远看不到本实例、TTL 永远过期。实装：`enableCluster` 中以 5s 周期调度心跳，shutdown 时随分区租约一并释放（§13.6.4 step 5 / G12）。
+
+**④ 与 §13.2.8 原设计的差异**
+
+| 原设计 | 实装 | 原因 |
+|--------|------|------|
+| gen 数字（metaGen+1 覆盖） | FencingToken（bootEpoch:counter）+ CAS | gen 覆盖是 read-then-write 非原子；token 排序天然单调且跨重启有效 |
+| 心跳 value 含 ownedPartitions+gen | 心跳 value = `<ts>\|<addr>\|<load>` | 分配表已在 /em/assignments/*，心跳只承担租约+负载上报 |
+| 实例间转发保订阅可达 | sticky：instanceUrl 固定订阅者 | 转发路径故障域大、时序复杂，删除（见 ①） |
+| LOAD_BALANCE_STICKY 独立模式 | 合并入 LOAD_BALANCE（partitionkey 路由） | sticky 成为唯一拓扑后无需独立模式 |
+
### 13.3 下发可靠性与消息语义
> **🔎 实现状态（v1.11 / 2026-07-06 盘点）**：⚠️ §13.3.1 ACK（offset 仅 ACK 推进）、§13.3.2 重试+DLQ（指数退避+`<topic>.DLQ`）、§13.3.5 去重声明、§13.3.6 不支持事务——均已实现（`ReliableDispatcher`）。**缺口**：§13.3.2 退避无 jitter（G13）；§13.3.3 STICKY 单实例✅但多实例退化为 RoundRobin（G8）；§13.3.4 TTL 过期丢弃未实现（附录 F.5）。
diff --git a/eventmesh-runtime/build.gradle b/eventmesh-runtime/build.gradle
index 4b8e0d4..222978c 100644
--- a/eventmesh-runtime/build.gradle
+++ b/eventmesh-runtime/build.gradle
@@ -110,8 +110,7 @@
'org.apache.eventmesh.runtime.it.StreamingSdkE2ETest',
'org.apache.eventmesh.runtime.it.RocketMQ5BrokerIntegrationTest',
'org.apache.eventmesh.runtime.it.RocketMQ5LiteHttpIntegrationTest',
- 'org.apache.eventmesh.runtime.it.KafkaClientE2EIntegrationTest',
- 'org.apache.eventmesh.runtime.it.NacosClusterForwardIntegrationTest'
+ 'org.apache.eventmesh.runtime.it.KafkaClientE2EIntegrationTest'
]
final List<String> BROKER_IT_4_CLASSES = [
'org.apache.eventmesh.runtime.it.RealBrokerIntegrationTest',
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshApplication.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshApplication.java
index 4561ea9..bfdd088 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshApplication.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshApplication.java
@@ -51,7 +51,7 @@
private org.apache.eventmesh.runtime.cluster.ClusterCoordinator clusterCoordinator;
private org.apache.eventmesh.runtime.cluster.ClusterMembership clusterMembership;
private org.apache.eventmesh.runtime.cluster.PartitionOwnership partitionOwnership;
- private org.apache.eventmesh.runtime.cluster.HttpForwarder httpForwarder;
+ private java.util.concurrent.ScheduledExecutorService heartbeatScheduler;
private String selfInstanceId;
private String advertisedAddr;
private javax.net.ssl.SSLContext sslContext;
@@ -147,14 +147,21 @@
/** Enable multi-instance coordination via a Meta-backed ClusterCoordinator (§13.2). */
public void enableCluster(org.apache.eventmesh.runtime.cluster.MetaStore metaStore, String selfInstanceId) {
this.selfInstanceId = selfInstanceId;
- // Full-sticky model (§3.1 / §5 stage 3): each instance pulls ALL partitions for the topics its
- // local subscribers need and delivers locally — NO cross-instance forwarding, NO partition
- // ownership assignment. Subscribers are pinned to one instance via the instanceUrl returned by
- // /events/subscribe (SDK poll+ack land on that instance). The cluster layer keeps only the
- // membership heartbeat (so /session/recommend can score instances globally by load).
+
+ // §13.2 cluster model: sticky delivery + partition fencing.
+ // - Each instance pulls partitions it OWNS (Meta CAS + fencing token, see PartitionOwnership)
+ // and delivers locally; no cross-instance forwarding.
+ // - Cross-instance forwarding (HttpForwarder / ClusterCoordinator forward path) is REMOVED in
+ // this release; subscribers are pinned to one instance via the instanceUrl from
+ // /events/subscribe so SDK poll+ack always land on the same instance.
+ // - Membership heartbeat keeps /session/recommend able to score instances globally by load.
+
+ org.apache.eventmesh.runtime.cluster.FencingToken selfToken =
+ new org.apache.eventmesh.runtime.cluster.FencingToken();
+
+ // 1. ClusterMembership — heartbeat value carries the fencing token + load snapshot.
this.clusterMembership = new org.apache.eventmesh.runtime.cluster.ClusterMembership(
- metaStore, selfInstanceId, selfInstanceId, 15_000L, System::currentTimeMillis);
- // Append the self-collected load snapshot to each heartbeat so /session/recommend can score.
+ metaStore, selfInstanceId, selfInstanceId, 15_000L, System::currentTimeMillis, selfToken);
org.apache.eventmesh.runtime.ingress.LoadMeter lm = runtime.ingress().loadMeter();
if (lm != null) {
this.clusterMembership.withLoadSupplier(() -> {
@@ -162,15 +169,29 @@
return lm.snapshot().toString();
});
}
- // PartitionOwnership + ClusterCoordinator/HttpForwarder are intentionally NOT wired: they
- // implemented the old "partition%n assign + cross-instance forward" broadcast model, which the
- // sticky model replaces. pullAndDispatch now pulls all partitions (ownedPartitions unset) and
- // delivers to local subscribers only. The classes are retained for an opt-in broadcast mode.
- // §13.6.3 dynamic config hot-reload: watch Meta for rate-limit rule changes.
+ // 2. Periodic heartbeat scheduler (fixes #5288: heartbeat was never scheduled, so
+ // /session/recommend never saw this instance).
+ this.heartbeatScheduler = java.util.concurrent.Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread t = new Thread(r, "em-heartbeat");
+ t.setDaemon(true);
+ return t;
+ });
+ heartbeatScheduler.scheduleAtFixedRate(
+ clusterMembership::heartbeat, 0, 5_000L, java.util.concurrent.TimeUnit.MILLISECONDS);
+
+ // 3. PartitionOwnership — wires CAS + fencing, drives ownedPartitions(topic) for the pull loop.
+ this.partitionOwnership = new org.apache.eventmesh.runtime.cluster.PartitionOwnership(
+ clusterMembership, metaStore, runtime.storage(), selfInstanceId,
+ 5_000L, System::currentTimeMillis, selfToken);
+ partitionOwnership.start(runtime.ingress()::activeTopicsClustered);
+ runtime.ingress().withPartitionOwnership(partitionOwnership);
+
+ // 4. Dynamic config hot-reload.
new org.apache.eventmesh.runtime.cluster.DynamicConfigWatcher(metaStore, runtime.ingress()).start();
- log.info("cluster enabled (sticky model): instance={} (membership + load heartbeat; no forwarding)", selfInstanceId);
+ log.info("cluster enabled (sticky + partition fencing): instance={} token={}",
+ selfInstanceId, selfToken);
}
/** Start runtime + traffic HTTP + admin HTTP. */
@@ -201,9 +222,6 @@
clusterMembership.setSelfAddress(forwardAddr);
httpServer.withClusterMembership(clusterMembership);
}
- if (selfInstanceId != null && httpForwarder != null) {
- httpServer.withCluster(selfInstanceId, httpForwarder);
- }
if (agentRegistrar != null) {
httpServer.withAgentRegistrar(agentRegistrar);
}
@@ -249,6 +267,9 @@
}
// §13.6.4 step 5 / G12: release the partition lease so peers re-assume ownership without
// waiting for the TTL (15s) to expire — minimises the handover gap on graceful shutdown.
+ if (heartbeatScheduler != null) {
+ heartbeatScheduler.shutdownNow();
+ }
if (partitionOwnership != null) {
partitionOwnership.stop();
}
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 4debf63..536cc55 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
@@ -184,6 +184,14 @@
}
/**
+ * The storage plugin this runtime boots - exposed for cluster wiring (PartitionOwnership's
+ * partitionCount / assignPartitions calls, 13.2.3).
+ */
+ public MeshStoragePlugin storage() {
+ return storage;
+ }
+
+ /**
* Pull-loop: poll each active topic from storage + dispatch to subscribers. Synchronized to
* prevent concurrent {@code storage.poll} calls on the same consumer (the 3-thread scheduler
* can otherwise overlap ticks when poll blocks, racing the consumer's internal state and losing
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/ClusterCoordinator.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/ClusterCoordinator.java
index 259cf56..3064c74 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/ClusterCoordinator.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/ClusterCoordinator.java
@@ -100,10 +100,11 @@
private List<ClusterSub> selectByMode(List<ClusterSub> targets, org.apache.eventmesh.common.wire.EventMeshFrame event) {
DistributionMode mode = targets.get(0).getMode();
switch (mode) {
- case LOAD_BALANCE_STICKY: {
+ case LOAD_BALANCE: {
// §13.3.3: stable hash(partitionkey) → one subscriber, so the same key always lands
// on the same worker across the whole cluster (order preserved). Sort by clientId
// first so every instance computes the same index for the same key/subscriber-set.
+ // When no partitionkey is present, fall back to round-robin.
java.util.List<ClusterSub> ordered = new java.util.ArrayList<>(targets);
ordered.sort(java.util.Comparator.comparing(ClusterSub::getClientId));
String key = event.attributes().get("partitionkey");
@@ -112,10 +113,6 @@
: Math.floorMod(key.hashCode(), ordered.size());
return java.util.Collections.singletonList(ordered.get(idx));
}
- case LOAD_BALANCE: {
- int idx = (roundRobin.getAndIncrement() & 0x7fffffff) % targets.size();
- return java.util.Collections.singletonList(targets.get(idx));
- }
case BROADCAST:
case MULTICAST:
default:
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/ClusterMembership.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/ClusterMembership.java
index 961814a..02f71ad 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/ClusterMembership.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/ClusterMembership.java
@@ -48,18 +48,22 @@
private volatile String selfAddress;
private final long ttlMs;
private final LongSupplier clock;
+ /** Per-JVM fencing token (§13.2.8④). Shared with PartitionOwnership for CAS assignment. */
+ private final FencingToken selfToken;
/** Optional load snapshot supplier (LoadMeter.sample()+snapshot()); null = no load in heartbeat. */
private volatile java.util.function.Supplier<String> loadSupplier;
/** Cached live set, refreshed on demand. */
private final ConcurrentHashMap<String, Boolean> liveCache = new ConcurrentHashMap<>();
- public ClusterMembership(MetaStore meta, String selfInstanceId, String selfAddress, long ttlMs, LongSupplier clock) {
+ public ClusterMembership(MetaStore meta, String selfInstanceId, String selfAddress, long ttlMs,
+ LongSupplier clock, FencingToken selfToken) {
this.meta = meta;
this.selfInstanceId = selfInstanceId;
this.selfAddress = selfAddress;
this.ttlMs = ttlMs;
this.clock = clock;
+ this.selfToken = selfToken;
}
/**
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/FencingToken.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/FencingToken.java
new file mode 100644
index 0000000..6e63d14
--- /dev/null
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/FencingToken.java
@@ -0,0 +1,107 @@
+/*
+ * 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.cluster;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Monotonic fencing token for partition ownership (§13.2.8④).
+ *
+ * <p>Each EventMesh instance generates a token at JVM start: {@code bootEpoch + ":" + counter}.
+ * The {@code bootEpoch} is {@code System.currentTimeMillis()} captured at construction; the
+ * {@code counter} is incremented on every {@link #next()} call. Tokens are ordered first by
+ * {@code bootEpoch} (older JVMs always lose), then by {@code counter} within the same epoch.</p>
+ *
+ * <p>A stale owner whose token is lower than the current Meta value is fenced and must stop
+ * polling that partition. The token survives process restarts because it is persisted in Meta
+ * (the value of {@code /em/assignments/<topic#partition>}).</p>
+ *
+ * <p>Thread-safety: {@link #next()} is safe to call from multiple threads. Each token's
+ * comparison value is an immutable snapshot taken at construction, so a token's ordering never
+ * changes after it is created — the shared counter only seeds future {@link #next()} calls.</p>
+ */
+public final class FencingToken implements Comparable<FencingToken> {
+
+ private final long bootEpoch;
+ /** Immutable comparison snapshot: the generator value captured at construction time. */
+ private final long value;
+ /** Shared monotonic counter; {@link #next()} increments it before snapshotting. */
+ private final AtomicLong counter;
+
+ public FencingToken() {
+ this(System.currentTimeMillis(), new AtomicLong(0));
+ }
+
+ FencingToken(long bootEpoch, AtomicLong counter) {
+ this.bootEpoch = bootEpoch;
+ this.counter = counter;
+ this.value = counter.get();
+ }
+
+ /**
+ * Allocate the next strictly-greater token.
+ *
+ * <p>Increments the shared counter and returns a token snapshotting the new value. The
+ * returned token compares greater than this token (and every token previously returned by
+ * this generator), while this token's own comparison value stays fixed at its
+ * construction-time snapshot.</p>
+ */
+ public FencingToken next() {
+ counter.incrementAndGet();
+ return new FencingToken(bootEpoch, counter);
+ }
+
+ @Override
+ public int compareTo(FencingToken o) {
+ if (this.bootEpoch != o.bootEpoch) {
+ return Long.compare(this.bootEpoch, o.bootEpoch);
+ }
+ return Long.compare(this.value, o.value);
+ }
+
+ @Override
+ public String toString() {
+ return bootEpoch + ":" + value;
+ }
+
+ public long bootEpoch() {
+ return bootEpoch;
+ }
+
+ /**
+ * Parse a token from its {@link #toString()} form.
+ *
+ * @throws IllegalArgumentException if {@code s} is not a {@code "<long>:<long>"} pair
+ */
+ public static FencingToken parse(String s) {
+ if (s == null) {
+ throw new IllegalArgumentException("token must not be null");
+ }
+ int sep = s.indexOf(':');
+ if (sep < 0) {
+ throw new IllegalArgumentException("malformed token (missing ':'): " + s);
+ }
+ try {
+ long epoch = Long.parseLong(s.substring(0, sep));
+ long count = Long.parseLong(s.substring(sep + 1));
+ return new FencingToken(epoch, new AtomicLong(count));
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException("malformed token (non-numeric): " + s, e);
+ }
+ }
+}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/HttpForwarder.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/HttpForwarder.java
deleted file mode 100644
index 523b2ff..0000000
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/HttpForwarder.java
+++ /dev/null
@@ -1,122 +0,0 @@
-/*
- * 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.cluster;
-
-import java.io.OutputStream;
-import java.net.HttpURLConnection;
-import java.net.URL;
-
-import io.cloudevents.CloudEvent;
-import io.cloudevents.core.provider.EventFormatProvider;
-import io.cloudevents.jackson.JsonFormat;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ObjectNode;
-
-import lombok.extern.slf4j.Slf4j;
-
-/**
- * Cross-instance forwarding over HTTP (§13.2.5 / §17.6). When the partition owner pulls a message
- * whose subscriber lives on another instance, this forwards it via {@code POST /internal/forward};
- * a late reply whose requestor lives elsewhere is forwarded via {@code POST /internal/reply-forward}.
- * Target instance addresses come from {@link ClusterMembership#addressOf} (the heartbeat value).
- *
- * <p>Synchronous HttpURLConnection — forwarding is on the dispatch hot path but each forward is one
- * short HTTP POST; the caller (ClusterCoordinator) treats a failed forward as a non-delivery and
- * the reliability layer retries/redelivers as needed.</p>
- */
-@Slf4j
-public class HttpForwarder implements Forwarder {
-
- private final ClusterMembership membership;
- private final ObjectMapper mapper = new ObjectMapper();
-
- public HttpForwarder(ClusterMembership membership) {
- this.membership = membership;
- }
-
- @Override
- public boolean forward(String targetInstance, String clientId, String topic, org.apache.eventmesh.common.wire.EventMeshFrame event) {
- String address = membership.addressOf(targetInstance);
- if (address == null) {
- log.warn("forward: no address for instance {}, dropping", targetInstance);
- return false;
- }
- return post("http://" + address + "/internal/forward", buildForwardBody(clientId, topic, event));
- }
-
- /** Forward a reply to the instance that issued the request (§17.6 self-addressed routing). */
- public boolean forwardReply(String targetInstance, String correlationId, CloudEvent replyEvent) {
- String address = membership.addressOf(targetInstance);
- if (address == null) {
- log.warn("forwardReply: no address for instance {}, dropping", targetInstance);
- return false;
- }
- return post("http://" + address + "/internal/reply-forward", buildReplyBody(correlationId, replyEvent));
- }
-
- private byte[] buildForwardBody(String clientId, String topic, org.apache.eventmesh.common.wire.EventMeshFrame event) {
- try {
- // Egress: forward body is CloudEvents-JSON over HTTP; Frame → CE-JSON via FrameAdaptor SPI.
- byte[] eventJson = org.apache.eventmesh.protocol.api.FrameAdaptors.toCloudEventsJson(event);
- ObjectNode body = mapper.createObjectNode();
- body.put("clientId", clientId);
- body.put("topic", topic);
- body.set("event", mapper.readTree(eventJson));
- return mapper.writeValueAsBytes(body);
- } catch (Exception e) {
- throw new RuntimeException("build forward body failed", e);
- }
- }
-
- private byte[] buildReplyBody(String correlationId, CloudEvent replyEvent) {
- try {
- ObjectNode body = mapper.createObjectNode();
- body.put("correlationId", correlationId);
- byte[] eventJson = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE).serialize(replyEvent);
- body.set("event", mapper.readTree(eventJson));
- return mapper.writeValueAsBytes(body);
- } catch (Exception e) {
- throw new RuntimeException("build reply body failed", e);
- }
- }
-
- private boolean post(String url, byte[] payload) {
- try {
- HttpURLConnection conn = (HttpURLConnection) new URL(url).openConnection();
- conn.setRequestMethod("POST");
- conn.setDoOutput(true);
- conn.setRequestProperty("Content-Type", "application/json");
- conn.setConnectTimeout(3000);
- conn.setReadTimeout(5000);
- try (OutputStream os = conn.getOutputStream()) {
- os.write(payload);
- }
- int status = conn.getResponseCode();
- try {
- conn.getInputStream().close();
- } catch (Exception ignored) {
- // best-effort drain
- }
- return status >= 200 && status < 300;
- } catch (Exception e) {
- log.warn("POST {} failed: {}", url, e.toString());
- return false;
- }
- }
-}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/InMemoryMetaStore.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/InMemoryMetaStore.java
index 7d05438..1d1017d 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/InMemoryMetaStore.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/InMemoryMetaStore.java
@@ -79,6 +79,24 @@
return false;
}
+ @Override
+ public boolean tryAcquire(String key, String expectedOldValue, String newValue) {
+ if (expectedOldValue == null) {
+ // key must be absent → use putIfAbsent
+ if (kv.putIfAbsent(key, newValue) == null) {
+ notify(key, newValue, false);
+ return true;
+ }
+ return false;
+ }
+ // CAS on existing value — ConcurrentHashMap.replace(key, oldVal, newVal) is atomic
+ boolean ok = kv.replace(key, expectedOldValue, newValue);
+ if (ok) {
+ notify(key, newValue, false);
+ }
+ return ok;
+ }
+
private void notify(String key, String value, boolean deleted) {
for (Watch w : watches) {
if (key.startsWith(w.prefix)) {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/MetaStore.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/MetaStore.java
index f244510..e789ce7 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/MetaStore.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/MetaStore.java
@@ -53,4 +53,21 @@
* @return true if the key existed and was removed
*/
boolean delete(String key);
+
+ /**
+ * Atomic compare-and-set on a single key (§13.2.8④ fencing). Succeeds only when the current
+ * value equals {@code expectedOldValue} (or {@code expectedOldValue} is null and the key is
+ * absent); on success the key is set to {@code newValue} and {@code true} is returned.
+ *
+ * <p>On a false return the caller must re-read with {@link #get(String)} and decide whether
+ * to retry, fence itself, or give up. The Nacos ConfigService implementation uses
+ * {@code publishConfigCas(..., casMd5)}; the in-memory implementation uses
+ * {@code AtomicReference.compareAndSet}.</p>
+ *
+ * @param key target key
+ * @param expectedOldValue the value we expect to find (null = key absent)
+ * @param newValue the value to install if the expectation holds
+ * @return true on success, false on expectation mismatch or backend failure
+ */
+ boolean tryAcquire(String key, String expectedOldValue, String newValue);
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/NacosMetaStore.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/NacosMetaStore.java
index 26d0fc6..d573c40 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/NacosMetaStore.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/NacosMetaStore.java
@@ -49,10 +49,10 @@
* carries instanceId/timestamp/address); {@code getWithPrefix("/em/instances/")} lists all registered
* instances and reassembles the {@code timestamp|address} value; {@code delete} deregisters.</p>
*
- * <p><b>Limitations</b>: ConfigService still has no prefix-scan for non-instance prefixes, and
- * {@code putIfAbsent} is read-then-write (non-atomic) — neither affects multi-instance correctness
- * (instance discovery is the only prefix-scan consumer, and it now goes through NamingService).
- * This impl is compile-verified; runtime verification needs a live Nacos server.</p>
+ * <p><b>Limitations</b>: ConfigService has no prefix-scan for non-instance prefixes; {@code putIfAbsent}
+ * is read-then-write (non-atomic). Partition ownership fencing uses {@link #tryAcquire} (Nacos 2.x
+ * {@code publishConfigCas}), which is a true atomic CAS. This impl is compile-verified; runtime
+ * verification needs a live Nacos server.</p>
*/
@Slf4j
public class NacosMetaStore implements MetaStore {
@@ -146,8 +146,9 @@
@Override
public boolean putIfAbsent(String key, String value) {
- // ConfigService has no CAS — read-then-write (non-atomic; document the race).
- // For instance keys, registerInstance is idempotent-overwrite (last writer wins), acceptable.
+ // ConfigService has no CAS — read-then-write (non-atomic). For atomic acquire use
+ // {@link #tryAcquire} (publishConfigCas). putIfAbsent is acceptable for instance/sub keys
+ // (NamingService registerInstance is idempotent last-writer-wins).
if (get(key) != null) {
return false;
}
@@ -221,6 +222,47 @@
}
}
+ @Override
+ public boolean tryAcquire(String key, String expectedOldValue, String newValue) {
+ if (isInstanceKey(key) || isSubKey(key)) {
+ // NamingService doesn't support CAS; fall back to plain put (last-writer-wins).
+ // Instance/sub keys are heartbeats and registrations — not fencing-critical.
+ put(key, newValue);
+ return true;
+ }
+ try {
+ // Nacos 2.x publishConfigCas(dataId, group, content, casMd5): the server rejects the
+ // publish unless its current content MD5 matches casMd5. casMd5 = MD5 of the expected
+ // current content (MD5("") for "key absent"), giving us a true atomic CAS.
+ String expectedContent = expectedOldValue == null ? "" : expectedOldValue;
+ String casMd5 = md5Hex(expectedContent);
+ boolean ok = config.publishConfigCas(dataId(key), GROUP, newValue, casMd5);
+ if (ok) {
+ knownKeys.add(key);
+ }
+ return ok;
+ } catch (NacosException e) {
+ log.warn("nacos tryAcquire (publishConfigCas) failed for {}: {}", key, e.toString());
+ return false;
+ }
+ }
+
+ /** MD5 hex digest (lowercase, 32 chars). Nacos casMd5 is the MD5 of the expected content. */
+ private static String md5Hex(String input) {
+ try {
+ java.security.MessageDigest md = java.security.MessageDigest.getInstance("MD5");
+ byte[] digest = md.digest(input.getBytes(java.nio.charset.StandardCharsets.UTF_8));
+ StringBuilder sb = new StringBuilder(32);
+ for (byte b : digest) {
+ sb.append(Character.forDigit((b >> 4) & 0xF, 16));
+ sb.append(Character.forDigit(b & 0xF, 16));
+ }
+ return sb.toString();
+ } catch (java.security.NoSuchAlgorithmException e) {
+ throw new RuntimeException("MD5 not available", e);
+ }
+ }
+
/**
* Nacos ConfigService dataIds reject {@code /}, {@code #} and several other path chars with
* "dataId invalid". The MetaStore key space uses slash-delimited paths
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/PartitionOwnership.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/PartitionOwnership.java
index 16d8fcb..af610dc 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/PartitionOwnership.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/PartitionOwnership.java
@@ -47,11 +47,14 @@
* reads {@link #ownedPartitions(String)} to decide which partitions to poll. When ownership is
* unknown (partitionCount -1, e.g. RocketMQ) it returns {@code null} → poll-all fallback.</p>
*
- * <p><b>Not yet here</b>: generation fencing (§13.2.8④, G3) and remote offset (§13.2.4, G5). Today
- * this is the "self-allocated" plan — every instance computes the same deterministic map, so no
- * instance overlaps, but a network-partitionned stale owner isn't fenced until its lease expires.
- * Nacos's lack of prefix-scan can also make {@code liveInstances} incomplete — etcd backend (G6)
- * fixes that.</p>
+ * <p><b>Fencing</b>: ownership is recorded in Meta via atomic CAS ({@link MetaStore#tryAcquire})
+ * with a monotonically increasing {@link FencingToken} (§13.2.8④). A stale owner whose token is
+ * lower than the current Meta value is fenced on its next refresh and stops polling that partition.
+ * Liveness is preserved across membership churn: partitions that leave our assigner share are
+ * released (CAS to a {@code ""} tombstone), and partitions held by a TTL-evicted owner are taken
+ * over regardless of token order (a partitioned zombie has already lost its own lease gate).
+ * Remote offset (§13.2.4, G5) is not yet here — the local offset store remains the source of truth
+ * for delivery progress.</p>
*/
@Slf4j
public class PartitionOwnership {
@@ -62,9 +65,11 @@
private final String selfInstanceId;
private final long intervalMs;
private final LongSupplier clock;
+ /** Per-JVM fencing token (§13.2.8④). next() produces strictly-increasing tokens for CAS. */
+ private final FencingToken fencingToken;
private final ConcurrentHashMap<String, List<Integer>> owned = new ConcurrentHashMap<>();
- private final ConcurrentHashMap<String, Long> myGen = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, FencingToken> myGen = new ConcurrentHashMap<>();
private final AtomicBoolean running = new AtomicBoolean(false);
/**
* Lease flag: true while the last heartbeat reached Meta. When false (Meta unreachable),
@@ -74,15 +79,18 @@
private volatile boolean leaseValid = true;
private ScheduledExecutorService scheduler;
private Supplier<Set<String>> topicSource;
+ /** Tombstone value for released assignment records (§13.2.10; empty string). */
+ private static final String RELEASED = "";
public PartitionOwnership(ClusterMembership membership, MetaStore metaStore, MeshStoragePlugin storage,
- String selfInstanceId, long intervalMs, LongSupplier clock) {
+ String selfInstanceId, long intervalMs, LongSupplier clock, FencingToken fencingToken) {
this.membership = membership;
this.metaStore = metaStore;
this.storage = storage;
this.selfInstanceId = selfInstanceId;
this.intervalMs = intervalMs;
this.clock = clock;
+ this.fencingToken = fencingToken;
}
/**
@@ -103,6 +111,16 @@
}
private void refresh() {
+ Set<String> topics = topicSource == null ? Collections.emptySet() : topicSource.get();
+ refreshOnce(topics);
+ }
+
+ /**
+ * One ownership cycle, split out of {@link #refresh()} for deterministic tests: the scheduler
+ * drives {@code refresh()} (which pulls the topic set from {@code topicSource}); tests call
+ * this directly with an explicit topic set. Package-private.
+ */
+ void refreshOnce(Set<String> topics) {
try {
// §13.2.8② lease = heartbeat. If Meta is unreachable the heartbeat returns false: lose
// the lease and skip this cycle. ownedPartitions() then returns empty (poll nothing),
@@ -113,8 +131,7 @@
return;
}
leaseValid = true;
- Set<String> topics = topicSource == null ? Collections.emptySet() : topicSource.get();
- if (topics.isEmpty()) {
+ if (topics == null || topics.isEmpty()) {
return;
}
List<String> live = membership.liveInstances();
@@ -129,12 +146,15 @@
}
Map<Integer, String> assignment = PartitionAssigner.assign(count, live);
List<Integer> mine = PartitionAssigner.ownedBy(assignment, selfInstanceId);
- // §13.2.8④ gen fencing: keep only partitions whose Meta assignment still says we're
- // the owner (or that we newly acquire). A stale owner whose lease expired sees a
- // newer gen in Meta and backs off (soft fencing — see class javadoc).
+ // §13.2.8④ fencing: keep only partitions whose Meta assignment still says we're
+ // the owner (or that we newly acquire via CAS). A stale owner whose lease expired
+ // sees a newer token in Meta and backs off (atomic CAS fencing — see class javadoc).
+ // Release partitions that left our assigner share (membership churn) so the new
+ // rightful owner is not fenced by our — possibly higher — token forever.
+ releaseStale(topic, mine);
List<Integer> fenced = new java.util.ArrayList<>(mine.size());
for (int p : mine) {
- if (acquireOrFence(topic, p)) {
+ if (acquireOrFence(topic, p, live)) {
fenced.add(p);
}
}
@@ -151,54 +171,130 @@
}
/**
- * Acquire (or confirm) ownership of one partition via the Meta assignment table, with a
- * monotonically increasing generation (§13.2.8④). Returns false when another instance has a
- * newer generation in Meta — we've been fenced and must stop polling this partition.
+ * Acquire (or confirm) ownership of one partition via the Meta assignment table, using an
+ * atomic CAS ({@link MetaStore#tryAcquire}) with a monotonically increasing {@link FencingToken}
+ * (§13.2.8④). Returns false when another instance holds a fencing token that is ≥ ours — we've
+ * been fenced and must stop polling this partition.
*
- * <p>Meta record: {@code /em/assignments/<topic#partition> = "<gen>|<ownerInstanceId>"}.</p>
+ * <p>Meta record: {@code /em/assignments/<topic#partition> = "<token>|<ownerInstanceId>"}.
+ *
+ * <p>The CAS replaces the old read-then-write race: two instances that both read {@code null}
+ * would both {@code put} and the last writer would silently win. With {@code tryAcquire},
+ * exactly one CAS succeeds and the loser re-reads on the next refresh cycle.</p>
*/
- private boolean acquireOrFence(String topic, int partition) {
+ private boolean acquireOrFence(String topic, int partition, List<String> live) {
String key = "/em/assignments/" + topic + "#" + partition;
String pkey = topic + "#" + partition;
- long metaGen = -1L;
- String metaOwner = null;
+
+ String currentRec = null;
+ FencingToken currentToken = null;
+ String currentOwner = null;
try {
- String rec = metaStore.get(key);
- if (rec != null) {
- int sep = rec.indexOf('|');
+ currentRec = metaStore.get(key);
+ if (currentRec != null) {
+ int sep = currentRec.indexOf('|');
if (sep > 0) {
- metaGen = Long.parseLong(rec.substring(0, sep));
- metaOwner = rec.substring(sep + 1);
+ currentToken = FencingToken.parse(currentRec.substring(0, sep));
+ currentOwner = currentRec.substring(sep + 1);
}
}
} catch (Exception e) {
log.debug("assignment read failed for {}: {}", key, e.toString());
}
- if (metaOwner == null) {
- // First claim.
- metaStore.put(key, "0|" + selfInstanceId);
- myGen.put(pkey, 0L);
- return true;
- }
- if (metaOwner.equals(selfInstanceId)) {
- // Still ours — keep our gen in sync with Meta (may have been refreshed by a restart).
- myGen.put(pkey, metaGen);
- return true;
- }
- // Another instance holds the Meta assignment.
- long mine = myGen.getOrDefault(pkey, -1L);
- if (metaGen > mine) {
- // They have a newer generation — we're fenced. Drop ownership.
- log.info("fenced: partition {}#{} taken over by {} (gen {} > our {})", topic, partition, metaOwner, metaGen, mine);
- myGen.remove(pkey);
+ FencingToken myToken = fencingToken.next();
+
+ // Case 1: unclaimed (absent, or the "" tombstone left by releaseStale) — CAS → our token
+ if (currentOwner == null) {
+ boolean ok = metaStore.tryAcquire(key, currentRec, myToken + "|" + selfInstanceId);
+ if (ok) {
+ myGen.put(pkey, myToken);
+ return true;
+ }
+ // Lost the race — another instance claimed it. Re-read next cycle.
return false;
}
- // We're (re)claiming it — bump the generation so any stale owner fences on its next refresh.
- long newGen = metaGen + 1;
- metaStore.put(key, newGen + "|" + selfInstanceId);
- myGen.put(pkey, newGen);
- return true;
+
+ // Case 2: still ours — sync local token with Meta
+ if (currentOwner.equals(selfInstanceId)) {
+ if (currentToken != null) {
+ myGen.put(pkey, currentToken);
+ }
+ return true;
+ }
+
+ // Case 3: another instance holds it — take over when their lease is gone (TTL eviction)
+ // or our token is strictly higher; otherwise we're fenced and stop polling.
+ boolean ownerEvicted = !live.contains(currentOwner);
+ FencingToken mine = myGen.get(pkey);
+ if (mine == null) {
+ mine = myToken;
+ }
+ if (ownerEvicted || currentToken == null || mine.compareTo(currentToken) > 0) {
+ // Fence them: CAS currentValue → our value. An evicted owner that is still polling has
+ // already failed its own heartbeat gate (leaseValid=false → polls nothing), so forcing
+ // the takeover is safe; a live higher-token owner keeps the partition.
+ boolean ok = metaStore.tryAcquire(key, currentRec, myToken + "|" + selfInstanceId);
+ if (ok) {
+ myGen.put(pkey, myToken);
+ return true;
+ }
+ // CAS failed — someone else changed it; re-read next cycle
+ return false;
+ }
+ // Their token is ≥ ours — we're fenced
+ log.info("fenced: partition {}#{} held by {} (token {} >= our {})",
+ topic, partition, currentOwner, currentToken, mine);
+ myGen.remove(pkey);
+ return false;
+ }
+
+ /**
+ * Release Meta assignment records for partitions of {@code topic} that are no longer in
+ * {@code mine} — the deterministic assigner moved them to a peer after membership churn. The
+ * record is CAS'd to the released tombstone {@code ""} (only if it still names us, so a
+ * concurrent takeover is never clobbered), letting the new rightful owner claim it on its next
+ * cycle instead of being fenced by our — possibly higher — token forever.
+ */
+ private void releaseStale(String topic, List<Integer> mine) {
+ String prefix = topic + "#";
+ for (String pkey : myGen.keySet()) {
+ if (!pkey.startsWith(prefix)) {
+ continue;
+ }
+ int p;
+ try {
+ p = Integer.parseInt(pkey.substring(prefix.length()));
+ } catch (NumberFormatException e) {
+ continue;
+ }
+ if (mine.contains(p)) {
+ continue;
+ }
+ String key = "/em/assignments/" + pkey;
+ String rec = null;
+ try {
+ rec = metaStore.get(key);
+ } catch (Exception e) {
+ log.debug("assignment read failed for {}: {}", key, e.toString());
+ }
+ if (selfInstanceId.equals(ownerOf(rec))) {
+ try {
+ metaStore.tryAcquire(key, rec, RELEASED);
+ } catch (Exception e) {
+ log.debug("assignment release failed for {}: {}", key, e.toString());
+ }
+ }
+ myGen.remove(pkey);
+ }
+ }
+
+ private static String ownerOf(String record) {
+ if (record == null || record.isEmpty()) {
+ return null;
+ }
+ int sep = record.indexOf('|');
+ return sep > 0 ? record.substring(sep + 1) : null;
}
/**
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/http/UniHttpServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/http/UniHttpServer.java
index 7e9352b..e67bdd5 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/http/UniHttpServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/http/UniHttpServer.java
@@ -77,7 +77,6 @@
private org.apache.eventmesh.runtime.security.FilterChain filterChain;
private org.apache.eventmesh.runtime.transport.http.LegacyHttpBridge legacyBridge;
private String selfInstanceId;
- private org.apache.eventmesh.runtime.cluster.HttpForwarder forwarder;
private org.apache.eventmesh.runtime.session.AgentRegistrar agentRegistrar;
private org.apache.eventmesh.runtime.session.Matchmaker matchmaker;
private org.apache.eventmesh.runtime.session.SessionRouter sessionRouter;
@@ -128,16 +127,8 @@
}
/**
- * Wire cross-instance forwarding (§13.2.5 / §17.6). {@code selfInstanceId} identifies this
- * instance (for self-addressed reply routing); {@code forwarder} does the HTTP POST to peers.
+ * Wire cluster membership so {@code /session/recommend} can read live instances + load (§3.2).
*/
- public UniHttpServer withCluster(String selfInstanceId, org.apache.eventmesh.runtime.cluster.HttpForwarder forwarder) {
- this.selfInstanceId = selfInstanceId;
- this.forwarder = forwarder;
- return this;
- }
-
- /** Wire cluster membership so {@code /session/recommend} can read live instances + load (§3.2). */
public UniHttpServer withClusterMembership(org.apache.eventmesh.runtime.cluster.ClusterMembership membership) {
this.clusterMembership = membership;
return this;
@@ -226,8 +217,6 @@
server.createContext("/session/stream", this::sessionStream);
server.createContext("/session/publish", this::sessionPublish);
server.createContext("/session/subscribe", this::sessionSubscribe);
- server.createContext("/internal/forward", this::forwardInternal);
- server.createContext("/internal/reply-forward", this::replyForwardInternal);
if (legacyBridge != null) {
server.createContext("/eventmesh/publish", this::legacyPublish);
server.createContext("/eventmesh/subscribe", this::legacySubscribe);
@@ -522,58 +511,17 @@
String corrId = text(body, "correlationId");
CloudEvent replyEvent = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE)
.deserialize(mapper.writeValueAsBytes(body.get("event")));
- // §17.6 self-addressed routing: if the requestor lives on another instance, forward there.
- Object replyInst = replyEvent.getExtension("emreplyinstance");
- if (replyInst != null && !replyInst.toString().equals(selfInstanceId) && forwarder != null) {
- boolean ok = forwarder.forwardReply(replyInst.toString(), corrId, replyEvent);
- writeJson(exchange, ok ? 200 : 502, ack(ok ? "forwarded" : "forward failed"));
- return;
- }
+ // §17.6 reply routing (sticky model - no cross-instance forwarding).
+ // Cross-instance reply forwarding is REMOVED with the forward path: the client posts
+ // the reply to the instance it sent the request to (pinned via instanceUrl); a reply
+ // landing on the wrong instance 404s (unknown correlationId) and the caller retries
+ // on the correct instance.
writeJson(exchange, ingress.reply(corrId, replyEvent) ? 200 : 404, ack("ok"));
} catch (Exception e) {
writeJson(exchange, 500, error("reply error: " + e.getMessage()));
}
}
- /** Cross-instance message forward (§13.2.5): peer pulled a message whose subscriber is here. */
- private void forwardInternal(HttpExchange exchange) throws IOException {
- if (!"POST".equals(exchange.getRequestMethod())) {
- writeJson(exchange, 405, error("method not allowed"));
- return;
- }
- try {
- JsonNode body = readJson(exchange);
- String clientId = text(body, "clientId");
- String topic = text(body, "topic");
- CloudEvent event = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE)
- .deserialize(mapper.writeValueAsBytes(body.get("event")));
- // Ingress boundary: the forward arrived as CloudEvents-JSON (HTTP wire); convert to the
- // internal EventMeshFrame before delivering locally.
- boolean ok = ingress.deliverLocal(topic, clientId,
- org.apache.eventmesh.common.wire.EventMeshFrame.fromCloudEvent(event));
- writeJson(exchange, ok ? 200 : 404, ack(ok ? "delivered" : "no local subscriber"));
- } catch (Exception e) {
- writeJson(exchange, 500, error("forward error: " + e.getMessage()));
- }
- }
-
- /** Cross-instance reply forward (§17.6): peer received a reply whose requestor is here. */
- private void replyForwardInternal(HttpExchange exchange) throws IOException {
- if (!"POST".equals(exchange.getRequestMethod())) {
- writeJson(exchange, 405, error("method not allowed"));
- return;
- }
- try {
- JsonNode body = readJson(exchange);
- String corrId = text(body, "correlationId");
- CloudEvent replyEvent = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE)
- .deserialize(mapper.writeValueAsBytes(body.get("event")));
- writeJson(exchange, ingress.reply(corrId, replyEvent) ? 200 : 404, ack("ok"));
- } catch (Exception e) {
- writeJson(exchange, 500, error("reply-forward error: " + e.getMessage()));
- }
- }
-
private void stream(HttpExchange exchange) throws IOException {
// SSE: hold the response open and pump buffered events to the client (§5). Blocks this
// thread until the client disconnects; Java-21 virtual threads (Phase 7) will make this cheap.
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/subscription/DistributionMode.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/subscription/DistributionMode.java
index 9810708..0ce16c5 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/subscription/DistributionMode.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/subscription/DistributionMode.java
@@ -38,12 +38,5 @@
/**
* Each message is delivered to the subscribers whose {@link CloudEventFilter} matches it.
*/
- MULTICAST,
-
- /**
- * Like {@link #LOAD_BALANCE}, but a subscriber is chosen by hashing a partition key so that
- * messages with the same key always go to the same subscriber (ordering). Reserved for
- * Phase 5.5 (§13.3.3); behaves like {@link #LOAD_BALANCE} until the partition-key wiring lands.
- */
- LOAD_BALANCE_STICKY
+ MULTICAST
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/subscription/SubscriptionManager.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/subscription/SubscriptionManager.java
index 8df2cea..15812cb 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/subscription/SubscriptionManager.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/subscription/SubscriptionManager.java
@@ -39,10 +39,10 @@
* <p>Unlike the legacy model (which delegates distribution to a MQ consumer group), this component
* pulls CloudEvents from {@link MeshStoragePlugin#poll} and decides — per event, per the
* subscription rules — which subscribers receive it (§4). Distribution modes:
- * {@link DistributionMode#LOAD_BALANCE} (round-robin one subscriber),
+ * {@link DistributionMode#LOAD_BALANCE} (one subscriber — hash-of-partitionkey when present,
+ * round-robin otherwise, §13.3.3),
* {@link DistributionMode#BROADCAST} (all subscribers),
- * {@link DistributionMode#MULTICAST} (filter-matched subscribers),
- * {@link DistributionMode#LOAD_BALANCE_STICKY} (hash-of-partition-key one subscriber, §13.3.3).</p>
+ * {@link DistributionMode#MULTICAST} (filter-matched subscribers).</p>
*
* <p>Phase 2 scope: single-instance dispatch logic, push→pull wiring against the storage plugin,
* and heartbeat-based subscriber liveness. Multi-instance coordination (partition assignment,
@@ -289,6 +289,8 @@
// Route by the mode of the active set. When a topic is shared by subscribers in different
// modes, the mode of the first live subscriber wins (documented limitation for Phase 2;
// mixed-mode topics are not a target use case).
+ // §13.3.3: LOAD_BALANCE absorbs the former LOAD_BALANCE_STICKY behaviour — if the event
+ // carries a partitionkey attribute we hash into the same subscriber, otherwise round-robin.
switch (active.get(0).getMode()) {
case BROADCAST:
return active;
@@ -300,11 +302,9 @@
}
}
return matched;
- case LOAD_BALANCE_STICKY:
- return Collections.singletonList(active.get(stickyIndex(event, active.size())));
case LOAD_BALANCE:
default:
- return Collections.singletonList(active.get(nextIndex(active.size())));
+ return Collections.singletonList(active.get(stickyOrRoundRobin(event, active.size())));
}
}
@@ -317,15 +317,15 @@
}
}
- private int nextIndex(int size) {
- // Math.abs(Integer.MIN_VALUE) is negative; mask instead.
- return (roundRobinCounter.getAndIncrement() & 0x7fffffff) % size;
- }
-
- private int stickyIndex(org.apache.eventmesh.common.wire.EventMeshFrame event, int size) {
+ private int stickyOrRoundRobin(org.apache.eventmesh.common.wire.EventMeshFrame event, int size) {
+ // §13.3.3: hash(partitionkey) when present → stable routing (same key → same subscriber);
+ // otherwise round-robin so LOAD_BALANCE absorbs the former LOAD_BALANCE_STICKY behaviour.
String key = event.attributes().get("partitionkey");
- int hash = key == null ? nextIndex(size) : key.hashCode();
- return Math.floorMod(hash, size);
+ if (key == null) {
+ // Math.abs(Integer.MIN_VALUE) is negative; mask instead.
+ return (roundRobinCounter.getAndIncrement() & 0x7fffffff) % size;
+ }
+ return Math.floorMod(key.hashCode(), size);
}
private void removeInternal(Subscription sub) {
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/ClusterCoordinationBaseTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/ClusterCoordinationBaseTest.java
index 76b8912..e1467f8 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/ClusterCoordinationBaseTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/ClusterCoordinationBaseTest.java
@@ -50,8 +50,8 @@
void membershipHeartbeatAndTtlExpiry() {
AtomicLong clock = new AtomicLong(1_000L);
InMemoryMetaStore meta = new InMemoryMetaStore();
- ClusterMembership a = new ClusterMembership(meta, "A", "A", 5_000L, clock::get);
- ClusterMembership b = new ClusterMembership(meta, "B", "B", 5_000L, clock::get);
+ ClusterMembership a = new ClusterMembership(meta, "A", "A", 5_000L, clock::get, new FencingToken());
+ ClusterMembership b = new ClusterMembership(meta, "B", "B", 5_000L, clock::get, new FencingToken());
a.heartbeat();
b.heartbeat();
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/ClusterDeliveryFaultTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/ClusterDeliveryFaultTest.java
new file mode 100644
index 0000000..52c63c1
--- /dev/null
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/ClusterDeliveryFaultTest.java
@@ -0,0 +1,386 @@
+/*
+ * 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.cluster;
+
+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.storage.MeshStoragePlugin;
+import org.apache.eventmesh.common.wire.EventMeshFrame;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * In-process fault-injection tests for the unified delivery topology (§13.2.10, #5293).
+ *
+ * <p>Three (or four) instances share one {@link InMemoryMetaStore}; every ownership cycle is
+ * driven explicitly via {@link PartitionOwnership#refreshOnce} (no scheduler, no sleeps) against
+ * a mutable clock, so failures — crash, TTL expiry, Meta network partition, membership churn —
+ * are fully deterministic.</p>
+ *
+ * <p>Scenarios: steady-state deterministic split; crash takeover (TTL eviction forces the CAS
+ * regardless of token order); scale-out churn (released partitions are re-claimed, no
+ * stranding); Meta partition (lease gate stops polling → no split-brain duplicates); healed
+ * partition (no failover, the share is reclaimed untouched).</p>
+ */
+class ClusterDeliveryFaultTest {
+
+ private static final String TOPIC = "orders";
+ private static final int PARTITIONS = 6;
+ private static final long TTL_MS = 5_000L;
+ private static final String ASSIGNMENT_PREFIX = "/em/assignments/";
+
+ /** Mutable clock — tests advance time explicitly instead of sleeping. */
+ static final class Clock {
+
+ private volatile long now = 1_000L;
+
+ long get() {
+ return now;
+ }
+
+ void advance(long ms) {
+ now += ms;
+ }
+ }
+
+ /** In-memory storage stub recording the last {@code assignPartitions} view per topic. */
+ static final class FakeStorage implements MeshStoragePlugin {
+
+ final Map<String, List<Integer>> assigned = new ConcurrentHashMap<>();
+
+ @Override
+ public void init(Properties properties) {
+ }
+
+ @Override
+ public void send(String topic, EventMeshFrame frame, SendCallback callback) {
+ }
+
+ @Override
+ public List<EventMeshFrame> poll(String topic, int partition, long startOffset, int maxEvents, long timeoutMs) {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void assignPartitions(String topic, List<Integer> partitions) {
+ assigned.put(topic, new ArrayList<>(partitions));
+ }
+
+ @Override
+ public void commitOffset(String topic, int partition, long offset) {
+ }
+
+ @Override
+ public int partitionCount(String topic) {
+ return PARTITIONS;
+ }
+
+ @Override
+ public boolean isStarted() {
+ return true;
+ }
+
+ @Override
+ public boolean isClosed() {
+ return false;
+ }
+
+ @Override
+ public void start() {
+ }
+
+ @Override
+ public void shutdown() {
+ }
+ }
+
+ /** Wraps the shared Meta store; {@code writesFail = true} simulates a network partition from Meta. */
+ static final class PartitionedMetaStore implements MetaStore {
+
+ final MetaStore delegate;
+ volatile boolean writesFail;
+
+ PartitionedMetaStore(MetaStore delegate) {
+ this.delegate = delegate;
+ }
+
+ private void gate() {
+ if (writesFail) {
+ throw new RuntimeException("simulated Meta partition");
+ }
+ }
+
+ @Override
+ public void watch(String prefix, MetaListener listener) {
+ delegate.watch(prefix, listener);
+ }
+
+ @Override
+ public void put(String key, String value) {
+ gate();
+ delegate.put(key, value);
+ }
+
+ @Override
+ public boolean putIfAbsent(String key, String value) {
+ gate();
+ return delegate.putIfAbsent(key, value);
+ }
+
+ @Override
+ public String get(String key) {
+ return delegate.get(key);
+ }
+
+ @Override
+ public Map<String, String> getWithPrefix(String prefix) {
+ return delegate.getWithPrefix(prefix);
+ }
+
+ @Override
+ public boolean delete(String key) {
+ gate();
+ return delegate.delete(key);
+ }
+
+ @Override
+ public boolean tryAcquire(String key, String expectedOldValue, String newValue) {
+ gate();
+ return delegate.tryAcquire(key, expectedOldValue, newValue);
+ }
+ }
+
+ /** One simulated EventMesh instance: membership + partition ownership + storage view. */
+ static final class Instance {
+
+ final String id;
+ final PartitionedMetaStore meta;
+ final ClusterMembership membership;
+ final PartitionOwnership ownership;
+ final FakeStorage storage;
+
+ Instance(MetaStore sharedMeta, Clock clock, String id, long bootEpoch) {
+ this.id = id;
+ this.meta = new PartitionedMetaStore(sharedMeta);
+ FencingToken token = new FencingToken(bootEpoch, new AtomicLong(0));
+ this.membership = new ClusterMembership(meta, id, id + ":8080", TTL_MS, clock::get, token);
+ this.storage = new FakeStorage();
+ this.ownership = new PartitionOwnership(membership, meta, storage, id, 1_000L, clock::get, token);
+ }
+
+ void refresh() {
+ ownership.refreshOnce(Set.of(TOPIC));
+ }
+
+ List<Integer> owned() {
+ List<Integer> ps = ownership.ownedPartitions(TOPIC);
+ return ps == null ? Collections.emptyList() : ps;
+ }
+ }
+
+ private final InMemoryMetaStore meta = new InMemoryMetaStore();
+ private final Clock clock = new Clock();
+
+ private Instance newInstance(String id, long bootEpoch) {
+ return new Instance(meta, clock, id, bootEpoch);
+ }
+
+ private void runRounds(int rounds, Instance... instances) {
+ for (int i = 0; i < rounds; i++) {
+ for (Instance inst : instances) {
+ inst.refresh();
+ }
+ }
+ }
+
+ /** partition → owner per the {@code /em/assignments/*} records (tombstones skipped). */
+ private Map<Integer, String> metaAssignments() {
+ Map<Integer, String> out = new HashMap<>();
+ for (Map.Entry<String, String> e : meta.getWithPrefix(ASSIGNMENT_PREFIX).entrySet()) {
+ String value = e.getValue();
+ if (value == null || value.isEmpty()) {
+ continue; // released tombstone
+ }
+ int sep = value.indexOf('|');
+ if (sep <= 0) {
+ continue;
+ }
+ int p = Integer.parseInt(e.getKey().substring(e.getKey().lastIndexOf('#') + 1));
+ out.put(p, value.substring(sep + 1));
+ }
+ return out;
+ }
+
+ // ---- Scenario 1: steady state — deterministic, disjoint, full coverage ----
+
+ @Test
+ void steadyStateDeterministicSplit() {
+ Instance a = newInstance("A", 1000L);
+ Instance b = newInstance("B", 2000L);
+ Instance c = newInstance("C", 3000L);
+ runRounds(4, a, b, c);
+
+ // sorted [A, B, C], partition % 3
+ assertEquals(List.of(0, 3), a.owned());
+ assertEquals(List.of(1, 4), b.owned());
+ assertEquals(List.of(2, 5), c.owned());
+
+ // the storage layer received the same ownership view
+ assertEquals(List.of(0, 3), a.storage.assigned.get(TOPIC));
+ assertEquals(List.of(1, 4), b.storage.assigned.get(TOPIC));
+ assertEquals(List.of(2, 5), c.storage.assigned.get(TOPIC));
+
+ // Meta agrees on every partition, and no partition has two owners
+ Map<Integer, String> m = metaAssignments();
+ assertEquals(6, m.size(), "every partition has an assignment record");
+ assertEquals("A", m.get(0));
+ assertEquals("B", m.get(1));
+ assertEquals("C", m.get(2));
+ assertEquals("A", m.get(3));
+ assertEquals("B", m.get(4));
+ assertEquals("C", m.get(5));
+ }
+
+ // ---- Scenario 2: instance crash — TTL eviction forces takeover regardless of token order ----
+
+ @Test
+ void crashedInstancePartitionsAreTakenOver() {
+ Instance a = newInstance("A", 1000L);
+ Instance b = newInstance("B", 2000L);
+ Instance c = newInstance("C", 3000L);
+ runRounds(4, a, b, c);
+
+ // B crashes (no more heartbeat/refresh); its lease expires.
+ clock.advance(TTL_MS + 1_000L);
+ runRounds(4, a, c);
+
+ // Live set is now [A, C] even though A's token (1000) is LOWER than dead B's (2000) —
+ // eviction forces the CAS takeover, otherwise B's partitions would be stranded forever.
+ assertEquals(List.of(0, 2, 4), a.owned());
+ assertEquals(List.of(1, 3, 5), c.owned());
+
+ Map<Integer, String> m = metaAssignments();
+ assertFalse(m.containsValue("B"), "no partition may still name the crashed instance");
+ for (int p = 0; p < PARTITIONS; p++) {
+ assertEquals(p % 2 == 0 ? "A" : "C", m.get(p), "partition " + p + " covered after crash");
+ }
+ }
+
+ // ---- Scenario 3: scale-out churn — released partitions are re-claimed, no stranding ----
+
+ @Test
+ void membershipChurnMovesOnlyTheReassignedPartitions() {
+ Instance a = newInstance("A", 1000L);
+ Instance b = newInstance("B", 2000L);
+ Instance c = newInstance("C", 3000L);
+ runRounds(4, a, b, c);
+
+ // D joins: sorted [A, B, C, D], partition % 4 → A:0,4 B:1,5 C:2 D:3
+ Instance d = newInstance("D", 4000L);
+ runRounds(4, a, b, c, d);
+
+ assertEquals(List.of(0, 4), a.owned());
+ assertEquals(List.of(1, 5), b.owned());
+ assertEquals(List.of(2), c.owned());
+ assertEquals(List.of(3), d.owned());
+
+ // Without the release path A (token 1000) could never claim 4 from live B (token 2000):
+ // the partition would be stranded with no poller. Here B releases 4 to the tombstone and
+ // A claims it on a later cycle.
+ Map<Integer, String> m = metaAssignments();
+ assertEquals(6, m.size(), "no stranded partition after churn");
+ assertEquals("A", m.get(0));
+ assertEquals("B", m.get(1));
+ assertEquals("C", m.get(2));
+ assertEquals("D", m.get(3));
+ assertEquals("A", m.get(4));
+ assertEquals("B", m.get(5));
+ }
+
+ // ---- Scenario 4: Meta network partition — lease gate stops polling (split-brain guard) ----
+
+ @Test
+ void metaPartitionStopsPollingUntilTtlExpiry() {
+ Instance a = newInstance("A", 1000L);
+ Instance b = newInstance("B", 2000L);
+ Instance c = newInstance("C", 3000L);
+ runRounds(4, a, b, c);
+
+ // B is cut off from Meta: every write fails, heartbeat included.
+ b.meta.writesFail = true;
+ b.refresh();
+
+ // Lease invalid → B polls nothing (its 1,4 would otherwise duplicate the quorum's
+ // consumption once A and C take over after the TTL).
+ assertTrue(b.owned().isEmpty(), "partitioned instance must stop polling");
+
+ // While B's lease is still fresh, A and C keep their own shares only — availability is
+ // sacrificed for consistency until the lease expires.
+ runRounds(2, a, c);
+ assertEquals(List.of(0, 3), a.owned());
+ assertEquals(List.of(2, 5), c.owned());
+
+ // After TTL expiry the quorum covers B's partitions.
+ clock.advance(TTL_MS + 1_000L);
+ runRounds(4, a, c);
+ assertEquals(List.of(0, 2, 4), a.owned());
+ assertEquals(List.of(1, 3, 5), c.owned());
+ assertFalse(metaAssignments().containsValue("B"), "B's records were taken over");
+ }
+
+ // ---- Scenario 5: healed partition — no failover, share reclaimed untouched ----
+
+ @Test
+ void healedPartitionReclaimsItsPartitionsWithoutFailover() {
+ Instance a = newInstance("A", 1000L);
+ Instance b = newInstance("B", 2000L);
+ Instance c = newInstance("C", 3000L);
+ runRounds(4, a, b, c);
+
+ // B is briefly partitioned from Meta, then heals before its lease expires.
+ b.meta.writesFail = true;
+ b.refresh();
+ assertTrue(b.owned().isEmpty(), "while partitioned, B polls nothing");
+
+ b.meta.writesFail = false;
+ runRounds(2, a, b, c);
+
+ // No failover happened (B never expired), so every instance ends with exactly its own
+ // share — no spurious reassignment, no duplicate coverage.
+ assertEquals(List.of(0, 3), a.owned());
+ assertEquals(List.of(1, 4), b.owned());
+ assertEquals(List.of(2, 5), c.owned());
+ Map<Integer, String> m = metaAssignments();
+ assertEquals(6, m.size());
+ assertEquals("B", m.get(1));
+ assertEquals("B", m.get(4));
+ }
+}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/ClusterMembershipLoadTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/ClusterMembershipLoadTest.java
index 072a1d0..467a909 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/ClusterMembershipLoadTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/ClusterMembershipLoadTest.java
@@ -37,7 +37,7 @@
void heartbeatWritesLoadAndAddress() {
AtomicClock clock = new AtomicClock();
InMemoryMetaStore meta = new InMemoryMetaStore();
- ClusterMembership m = new ClusterMembership(meta, "self", "self:8080", 15_000L, clock::get);
+ ClusterMembership m = new ClusterMembership(meta, "self", "self:8080", 15_000L, clock::get, new FencingToken());
m.withLoadSupplier(() -> "3|2000|1500|0.25");
m.heartbeat();
@@ -50,7 +50,7 @@
void heartbeatWithoutLoadSupplierOmitsLoadFields() {
AtomicClock clock = new AtomicClock();
InMemoryMetaStore meta = new InMemoryMetaStore();
- ClusterMembership m = new ClusterMembership(meta, "self", "1.2.3.4:8080", 15_000L, clock::get);
+ ClusterMembership m = new ClusterMembership(meta, "self", "1.2.3.4:8080", 15_000L, clock::get, new FencingToken());
m.heartbeat();
assertEquals("0|1.2.3.4:8080", meta.get(ClusterMembership.INSTANCE_PREFIX + "self"));
}
@@ -63,7 +63,7 @@
meta.put("/em/instances/a", "19000|h1:8080|5|5000|4000|0.10"); // age 1000 < ttl
meta.put("/em/instances/b", "19500|h2:8080|20|5000000|4000000|0.90"); // heavy, age 500 < ttl
meta.put("/em/instances/c", "1000|h3:8080|1|100|50|0.01"); // age 19000 > ttl 15000 → pruned
- ClusterMembership m = new ClusterMembership(meta, "self", "self:8080", 15_000L, clock::get);
+ ClusterMembership m = new ClusterMembership(meta, "self", "self:8080", 15_000L, clock::get, new FencingToken());
Map<String, ClusterMembership.InstanceInfo> live = m.liveInstancesWithLoad();
assertEquals(2, live.size(), "stale peer c must be pruned");
@@ -77,7 +77,7 @@
@Test
void setSelfAddressOverridesPlaceholder() {
- ClusterMembership m = new ClusterMembership(new InMemoryMetaStore(), "self", "self", 15_000L, () -> 0);
+ ClusterMembership m = new ClusterMembership(new InMemoryMetaStore(), "self", "self", 15_000L, () -> 0, new FencingToken());
m.setSelfAddress("10.0.0.5:8080");
// addressOf(self) returns the overridden address.
assertEquals("10.0.0.5:8080", m.addressOf("self"));
@@ -88,7 +88,7 @@
AtomicClock clock = new AtomicClock(10_000L);
InMemoryMetaStore meta = new InMemoryMetaStore();
meta.put("/em/instances/old", "9500|h:8080"); // old-format peer, no load fields
- ClusterMembership m = new ClusterMembership(meta, "self", "self:8080", 15_000L, clock::get);
+ ClusterMembership m = new ClusterMembership(meta, "self", "self:8080", 15_000L, clock::get, new FencingToken());
Map<String, ClusterMembership.InstanceInfo> live = m.liveInstancesWithLoad();
assertEquals("h:8080", live.get("old").address);
assertNull(live.get("old").load, "peer without load fields must parse to null snapshot");
@@ -100,7 +100,7 @@
AtomicClock clock = new AtomicClock(10_000L);
InMemoryMetaStore meta = new InMemoryMetaStore();
meta.put("/em/instances/p", "9500|h:8080|7|300");
- ClusterMembership m = new ClusterMembership(meta, "self", "self:8080", 15_000L, clock::get);
+ ClusterMembership m = new ClusterMembership(meta, "self", "self:8080", 15_000L, clock::get, new FencingToken());
LoadMeter.Snapshot load = m.liveInstancesWithLoad().get("p").load;
assertEquals(7, load.activeSessions);
assertEquals(300L, load.inflowBytesPerSec);
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/FencingTokenTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/FencingTokenTest.java
new file mode 100644
index 0000000..9fe5f2f
--- /dev/null
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/FencingTokenTest.java
@@ -0,0 +1,106 @@
+/*
+ * 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.cluster;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for {@link FencingToken} (§13.2.8④).
+ */
+class FencingTokenTest {
+
+ @Test
+ void nextIsStrictlyGreater() {
+ FencingToken t = new FencingToken(1000L, new AtomicLong(0));
+ FencingToken t1 = t.next();
+ FencingToken t2 = t.next();
+ assertTrue(t1.compareTo(t) > 0, "next() > original");
+ assertTrue(t2.compareTo(t1) > 0, "second next() > first next()");
+ }
+
+ @Test
+ void differentBootEpochOrdersByEpochFirst() {
+ FencingToken old = new FencingToken(1000L, new AtomicLong(5));
+ FencingToken newer = new FencingToken(2000L, new AtomicLong(0));
+ assertTrue(newer.compareTo(old) > 0, "higher bootEpoch wins regardless of counter");
+ assertTrue(old.compareTo(newer) < 0);
+ }
+
+ @Test
+ void sameBootEpochOrdersByCounter() {
+ FencingToken a = new FencingToken(1000L, new AtomicLong(3));
+ FencingToken b = new FencingToken(1000L, new AtomicLong(7));
+ assertTrue(b.compareTo(a) > 0);
+ assertTrue(a.compareTo(b) < 0);
+ }
+
+ @Test
+ void equalTokensCompareAsZero() {
+ FencingToken a = new FencingToken(1000L, new AtomicLong(5));
+ FencingToken b = new FencingToken(1000L, new AtomicLong(5));
+ assertEquals(0, a.compareTo(b));
+ }
+
+ @Test
+ void toStringRoundTrip() {
+ FencingToken t = new FencingToken(1234567890L, new AtomicLong(42));
+ String s = t.toString();
+ assertEquals("1234567890:42", s);
+ FencingToken parsed = FencingToken.parse(s);
+ assertEquals(0, t.compareTo(parsed), "parse(toString()) should be equal");
+ }
+
+ @Test
+ void parseRejectsNull() {
+ assertThrows(IllegalArgumentException.class, () -> FencingToken.parse(null));
+ }
+
+ @Test
+ void parseRejectsMissingColon() {
+ assertThrows(IllegalArgumentException.class, () -> FencingToken.parse("noColon"));
+ }
+
+ @Test
+ void parseRejectsNonNumeric() {
+ assertThrows(IllegalArgumentException.class, () -> FencingToken.parse("abc:def"));
+ }
+
+ @Test
+ void defaultConstructorUsesCurrentTime() {
+ long before = System.currentTimeMillis();
+ FencingToken t = new FencingToken();
+ long after = System.currentTimeMillis();
+ assertTrue(t.bootEpoch() >= before, "bootEpoch >= before");
+ assertTrue(t.bootEpoch() <= after, "bootEpoch <= after");
+ }
+
+ @Test
+ void nextPreservesBootEpoch() {
+ FencingToken t = new FencingToken(5000L, new AtomicLong(0));
+ FencingToken t1 = t.next();
+ FencingToken t2 = t.next();
+ assertEquals(5000L, t1.bootEpoch());
+ assertEquals(5000L, t2.bootEpoch());
+ }
+}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/InMemoryMetaStoreTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/InMemoryMetaStoreTest.java
new file mode 100644
index 0000000..16775c7
--- /dev/null
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/InMemoryMetaStoreTest.java
@@ -0,0 +1,114 @@
+/*
+ * 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.cluster;
+
+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 java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for {@link InMemoryMetaStore#tryAcquire(String, String, String)} — the atomic CAS
+ * that backs partition fencing (§13.2.8④).
+ */
+class InMemoryMetaStoreTest {
+
+ @Test
+ void tryAcquireOnAbsentKeySucceeds() {
+ InMemoryMetaStore meta = new InMemoryMetaStore();
+ assertTrue(meta.tryAcquire("k1", null, "v1"));
+ assertEquals("v1", meta.get("k1"));
+ }
+
+ @Test
+ void tryAcquireAbsentFailsWhenKeyExists() {
+ InMemoryMetaStore meta = new InMemoryMetaStore();
+ meta.put("k1", "v1");
+ assertFalse(meta.tryAcquire("k1", null, "v2"));
+ assertEquals("v1", meta.get("k1"), "value must be unchanged on CAS failure");
+ }
+
+ @Test
+ void tryAcquireReplacesOnExactMatch() {
+ InMemoryMetaStore meta = new InMemoryMetaStore();
+ meta.put("k1", "v1");
+ assertTrue(meta.tryAcquire("k1", "v1", "v2"));
+ assertEquals("v2", meta.get("k1"));
+ }
+
+ @Test
+ void tryAcquireFailsOnValueMismatch() {
+ InMemoryMetaStore meta = new InMemoryMetaStore();
+ meta.put("k1", "v1");
+ assertFalse(meta.tryAcquire("k1", "wrong", "v2"));
+ assertEquals("v1", meta.get("k1"), "value must be unchanged on CAS failure");
+ }
+
+ @Test
+ void tryAcquireFailsOnNullExpectedButKeyPresent() {
+ InMemoryMetaStore meta = new InMemoryMetaStore();
+ meta.put("k1", "existing");
+ assertFalse(meta.tryAcquire("k1", null, "new"));
+ }
+
+ @Test
+ void tryAcquireIsAtomicUnderConcurrency() throws Exception {
+ InMemoryMetaStore meta = new InMemoryMetaStore();
+ int n = 20;
+ CountDownLatch start = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(n);
+ AtomicInteger winners = new AtomicInteger(0);
+
+ for (int i = 0; i < n; i++) {
+ final int idx = i;
+ Thread t = new Thread(() -> {
+ try {
+ start.await();
+ if (meta.tryAcquire("race-key", null, "v" + idx)) {
+ winners.incrementAndGet();
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ done.countDown();
+ }
+ });
+ t.setDaemon(true);
+ t.start();
+ }
+
+ start.countDown();
+ done.await();
+
+ assertEquals(1, winners.get(), "exactly one thread must win the CAS");
+ }
+
+ @Test
+ void tryAcquireNotifiesListenersOnChange() {
+ InMemoryMetaStore meta = new InMemoryMetaStore();
+ java.util.concurrent.atomic.AtomicReference<String> seen = new java.util.concurrent.atomic.AtomicReference<>();
+ meta.watch("/em/assignments/", (key, value, deleted) -> seen.set(value));
+
+ meta.tryAcquire("/em/assignments/test#0", null, "token1|instanceA");
+ assertEquals("token1|instanceA", seen.get());
+ }
+}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/LoadBalancingScoringTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/LoadBalancingScoringTest.java
index fc87895..b4822f1 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/LoadBalancingScoringTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/LoadBalancingScoringTest.java
@@ -63,7 +63,7 @@
// Instance C: stale (should be pruned)
meta.put("/em/instances/c", "1000|h3:8080|1|100|50|0.01");
- ClusterMembership m = new ClusterMembership(meta, "self", "self:8080", 15_000L, clock::get);
+ ClusterMembership m = new ClusterMembership(meta, "self", "self:8080", 15_000L, clock::get, new FencingToken());
Map<String, ClusterMembership.InstanceInfo> live = m.liveInstancesWithLoad();
assertEquals(2, live.size(), "stale instance c must be pruned");
@@ -88,7 +88,7 @@
// Overloaded instance (cpu > 0.8)
meta.put("/em/instances/overloaded", "9500|h2:8080|5|2000|1500|0.90");
- ClusterMembership m = new ClusterMembership(meta, "self", "self:8080", 15_000L, clock::get);
+ ClusterMembership m = new ClusterMembership(meta, "self", "self:8080", 15_000L, clock::get, new FencingToken());
Map<String, ClusterMembership.InstanceInfo> live = m.liveInstancesWithLoad();
LoadMeter.Snapshot normal = live.get("normal").load;
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/PartitionFencingTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/PartitionFencingTest.java
new file mode 100644
index 0000000..8e68d64
--- /dev/null
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/cluster/PartitionFencingTest.java
@@ -0,0 +1,168 @@
+/*
+ * 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.cluster;
+
+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 java.util.concurrent.atomic.AtomicLong;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests the CAS + FencingToken partition-ownership protocol (§13.2.8④).
+ *
+ * <p>These tests simulate the {@link PartitionOwnership#acquireOrFence} logic using
+ * {@link MetaStore#tryAcquire} + {@link FencingToken} directly, verifying the three critical
+ * scenarios: first claim, race (exactly-one-wins), and restart fencing.</p>
+ */
+class PartitionFencingTest {
+
+ private static final String ASSIGNMENT_KEY = "/em/assignments/orders#0";
+
+ // ---- Scenario 1: First claim (CAS null → token) ----
+
+ @Test
+ void firstClaimSucceedsViaCAS() {
+ InMemoryMetaStore meta = new InMemoryMetaStore();
+ FencingToken tokenA = new FencingToken(1000L, new AtomicLong(0));
+ FencingToken nextA = tokenA.next();
+
+ // Simulate acquireOrFence: key is absent → CAS(null, nextA|A)
+ boolean ok = meta.tryAcquire(ASSIGNMENT_KEY, null, nextA + "|instanceA");
+ assertTrue(ok, "first claim must succeed");
+ assertEquals(nextA + "|instanceA", meta.get(ASSIGNMENT_KEY));
+ }
+
+ // ---- Scenario 2: Two instances race for the same unclaimed partition ----
+
+ @Test
+ void exactlyOneWinsTheRace() {
+ InMemoryMetaStore meta = new InMemoryMetaStore();
+ FencingToken tokenA = new FencingToken(1000L, new AtomicLong(0));
+ FencingToken tokenB = new FencingToken(1001L, new AtomicLong(0));
+
+ FencingToken nextA = tokenA.next();
+ FencingToken nextB = tokenB.next();
+
+ // Both try to CAS null → their value
+ boolean firstWon = meta.tryAcquire(ASSIGNMENT_KEY, null, nextA + "|instanceA");
+ boolean secondWon = meta.tryAcquire(ASSIGNMENT_KEY, null, nextB + "|instanceB");
+
+ assertTrue(firstWon, "first CAS must succeed");
+ assertFalse(secondWon, "second CAS must fail (key already claimed)");
+ assertEquals(nextA + "|instanceA", meta.get(ASSIGNMENT_KEY));
+ }
+
+ // ---- Scenario 3: Restart fencing (new bootEpoch > old bootEpoch) ----
+
+ @Test
+ void restartedInstanceFencesStaleOwner() {
+ InMemoryMetaStore meta = new InMemoryMetaStore();
+
+ // Instance A (old) holds the partition with an old token.
+ FencingToken tokenA = new FencingToken(1000L, new AtomicLong(0));
+ FencingToken nextA = tokenA.next();
+ meta.put(ASSIGNMENT_KEY, nextA + "|instanceA");
+
+ // Instance A crashes and restarts → new bootEpoch, higher token.
+ FencingToken tokenANew = new FencingToken(2000L, new AtomicLong(0));
+ FencingToken nextANew = tokenANew.next();
+
+ // Read current value, compare tokens, CAS if ours is higher.
+ String currentRec = meta.get(ASSIGNMENT_KEY);
+ FencingToken currentToken = FencingToken.parse(currentRec.split("\\|", 2)[0]);
+ assertTrue(nextANew.compareTo(currentToken) > 0, "new bootEpoch must be higher");
+
+ boolean fenced = meta.tryAcquire(ASSIGNMENT_KEY, currentRec, nextANew + "|instanceA");
+ assertTrue(fenced, "restart fencing CAS must succeed");
+ assertEquals(nextANew + "|instanceA", meta.get(ASSIGNMENT_KEY));
+ }
+
+ // ---- Scenario 4: Stale owner with lower token is fenced ----
+
+ @Test
+ void lowerTokenCannotFenceHigherToken() {
+ InMemoryMetaStore meta = new InMemoryMetaStore();
+
+ // Instance B (newer bootEpoch) holds the partition.
+ FencingToken tokenB = new FencingToken(2000L, new AtomicLong(0));
+ FencingToken nextB = tokenB.next();
+ meta.put(ASSIGNMENT_KEY, nextB + "|instanceB");
+
+ // Instance A (older bootEpoch) tries to fence B → must fail.
+ FencingToken tokenA = new FencingToken(1000L, new AtomicLong(0));
+ FencingToken nextA = tokenA.next();
+
+ String currentRec = meta.get(ASSIGNMENT_KEY);
+ FencingToken currentToken = FencingToken.parse(currentRec.split("\\|", 2)[0]);
+
+ // Mirror acquireOrFence Case 3 exactly: the fencing CAS is only attempted when our
+ // token is strictly higher than the one in Meta. A's token is lower, so the guard
+ // fails and A is fenced without writing anything.
+ boolean fenced = false;
+ if (nextA.compareTo(currentToken) > 0) {
+ fenced = meta.tryAcquire(ASSIGNMENT_KEY, currentRec, nextA + "|instanceA");
+ }
+ assertFalse(fenced, "lower token must not fence higher token");
+ assertEquals(nextB + "|instanceB", meta.get(ASSIGNMENT_KEY), "value must be unchanged");
+ }
+
+ // ---- Scenario 5: Same instance reclaims (token still ours) ----
+
+ @Test
+ void sameInstanceReclaims() {
+ InMemoryMetaStore meta = new InMemoryMetaStore();
+
+ FencingToken tokenA = new FencingToken(1000L, new AtomicLong(0));
+ FencingToken nextA = tokenA.next();
+ meta.put(ASSIGNMENT_KEY, nextA + "|instanceA");
+
+ // Instance A reads back and finds it's still the owner — no CAS needed.
+ String currentRec = meta.get(ASSIGNMENT_KEY);
+ String owner = currentRec.split("\\|", 2)[1];
+ assertEquals("instanceA", owner, "still ours");
+ // Just sync local token; no write needed.
+ }
+
+ // ---- Scenario 6: CAS fails if value changed between read and write ----
+
+ @Test
+ void casFailsIfValueChangedBetweenReadAndWrite() {
+ InMemoryMetaStore meta = new InMemoryMetaStore();
+
+ FencingToken tokenA = new FencingToken(1000L, new AtomicLong(0));
+ FencingToken nextA = tokenA.next();
+ meta.put(ASSIGNMENT_KEY, nextA + "|instanceA");
+
+ // Read what we think is there.
+ String staleRead = meta.get(ASSIGNMENT_KEY);
+
+ // Another instance changes it underneath us.
+ FencingToken tokenB = new FencingToken(2000L, new AtomicLong(0));
+ FencingToken nextB = tokenB.next();
+ meta.put(ASSIGNMENT_KEY, nextB + "|instanceB");
+
+ // Our CAS with the stale expected value must fail.
+ FencingToken tokenC = new FencingToken(3000L, new AtomicLong(0));
+ FencingToken nextC = tokenC.next();
+ boolean ok = meta.tryAcquire(ASSIGNMENT_KEY, staleRead, nextC + "|instanceC");
+ assertFalse(ok, "CAS with stale expected value must fail");
+ }
+}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/ClusterForwardIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/ClusterForwardIntegrationTest.java
deleted file mode 100644
index f052184..0000000
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/ClusterForwardIntegrationTest.java
+++ /dev/null
@@ -1,225 +0,0 @@
-/*
- * 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.it;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-import org.apache.eventmesh.api.SendCallback;
-import org.apache.eventmesh.api.SendResult;
-import org.apache.eventmesh.api.storage.MeshStoragePlugin;
-import org.apache.eventmesh.common.wire.EventMeshFrame;
-import org.apache.eventmesh.runtime.admin.UniAdminService;
-import org.apache.eventmesh.runtime.cluster.ClusterCoordinator;
-import org.apache.eventmesh.runtime.cluster.ClusterMembership;
-import org.apache.eventmesh.runtime.cluster.ClusterSubscriptionStore;
-import org.apache.eventmesh.runtime.cluster.HttpForwarder;
-import org.apache.eventmesh.runtime.cluster.InMemoryMetaStore;
-import org.apache.eventmesh.runtime.cluster.MetaStore;
-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.push.BufferedEvent;
-
-import java.net.URI;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Queue;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.TimeUnit;
-
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Test;
-
-import io.cloudevents.CloudEvent;
-import io.cloudevents.core.builder.CloudEventBuilder;
-
-/**
- * Two-instance cluster integration test (§13.2): instance A and instance B share one {@link MetaStore}
- * and route events across instances via {@link ClusterCoordinator} + {@link HttpForwarder} over real
- * HTTP. A subscriber on A registers cluster-wide; a publisher on B publishes; B's coordinator sees
- * the subscriber lives on A and forwards via {@code POST /internal/forward}; A delivers locally; the
- * subscriber polls A and receives the event. In-memory storage stub — no broker.
- *
- * <p>Each instance's {@code selfInstanceId} is {@code localhost:<trafficPort>} so {@code addressOf}
- * resolves to a reachable HTTP address. Cluster-wide subscription is registered via
- * {@link ClusterCoordinator#subscribe} (the HTTP {@code /events/subscribe} path is local-only).</p>
- */
-class ClusterForwardIntegrationTest {
-
- private static final String TOPIC = "cross";
-
- private Instance instA;
- private Instance instB;
-
- @AfterEach
- void tearDown() {
- if (instA != null) {
- instA.close();
- }
- if (instB != null) {
- instB.close();
- }
- }
-
- @Test
- void publishOnBdeliversToSubscriberOnA() throws Exception {
- MetaStore meta = new InMemoryMetaStore();
- instA = boot(meta, "A"); // selfInstanceId set to localhost:<portA> inside boot
- instB = boot(meta, "B");
-
- // Subscriber c1 lives on A. ingress.subscribe now registers cluster-wide automatically
- // (fix: previously the HTTP /events/subscribe path was local-only and a peer's publish
- // never reached this subscriber).
- instA.ingress.subscribe(TOPIC, "c1", org.apache.eventmesh.runtime.subscription.DistributionMode.BROADCAST, null);
- // Heartbeat so the other instance can resolve A's HTTP address.
- instA.membership.heartbeat();
- instB.membership.heartbeat();
- Thread.sleep(200); // let the Meta writes settle
-
- // Publish on B + pull-and-dispatch: B's coordinator sees c1 is on A → HTTP-forward to A.
- CloudEvent event = CloudEventBuilder.v1()
- .withId("x1").withSource(URI.create("it")).withType("it.event").build();
- instB.ingress.publish(TOPIC, event).get(5, TimeUnit.SECONDS);
- instB.ingress.pullAndDispatch(TOPIC, 100, 0L);
-
- // A delivered locally (via /internal/forward → ingress.deliverLocal) → c1 polls A.
- List<BufferedEvent> received = new ArrayList<>();
- long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(3);
- while (received.isEmpty() && System.nanoTime() < deadline) {
- received.addAll(instA.ingress.poll("c1", 100, 100L));
- }
- assertEquals(1, received.size(), "event published on B should be forwarded to A and delivered to c1");
- assertEquals("x1", received.get(0).getEvent().attributes().get("id"));
- }
-
- /** Boot one instance: traffic HTTP server on port 0, cluster wired with selfInstanceId=address. */
- private Instance boot(MetaStore meta, String tag) throws Exception {
- MeshStoragePlugin storage = new InMemoryStorage();
- UniIngressService ingress = new UniIngressService(storage, new InMemoryOffsetStore());
- UniAdminService admin = new UniAdminService(ingress);
- UniHttpServer http = new UniHttpServer(ingress, admin);
- int port = http.start(0);
- String selfId = "localhost:" + port; // addressOf(selfInstanceId) returns this
-
- ClusterMembership membership = new ClusterMembership(meta, selfId, selfId, 15_000L, System::currentTimeMillis);
- HttpForwarder forwarder = new HttpForwarder(membership);
- ClusterSubscriptionStore subStore = new ClusterSubscriptionStore(meta);
- ClusterCoordinator coordinator = new ClusterCoordinator(selfId, subStore,
- (topic, clientId, event) -> {
- ingress.deliverLocal(topic, clientId, event);
- return true;
- }, forwarder);
- ingress.withCluster(coordinator);
- return new Instance(ingress, http, membership, coordinator, port);
- }
-
- private static final class Instance {
-
- final UniIngressService ingress;
- final UniHttpServer http;
- final ClusterMembership membership;
- final ClusterCoordinator coordinator;
- final int port;
-
- Instance(UniIngressService ingress, UniHttpServer http, ClusterMembership membership,
- ClusterCoordinator coordinator, int port) {
- this.ingress = ingress;
- this.http = http;
- this.membership = membership;
- this.coordinator = coordinator;
- this.port = port;
- }
-
- void close() {
- try {
- membership.leave();
- } catch (Exception ignored) {
- // best-effort
- }
- http.stop();
- }
- }
-
- // ---- in-memory storage (shared logical MQ; each instance has its own map but the test only
- // publishes on B and never polls storage on B, so a per-instance map is fine) ----
-
- static final class InMemoryStorage implements MeshStoragePlugin {
-
- private final ConcurrentHashMap<String, Queue<CloudEvent>> queues = new ConcurrentHashMap<>();
-
- @Override
- public void init(java.util.Properties p) {
- // no-op
- }
-
- @Override
- public void send(String topic, EventMeshFrame frame, SendCallback cb) {
- CloudEvent event = frame.toCloudEvent();
- queues.computeIfAbsent(topic, k -> new ConcurrentLinkedQueue<>()).offer(event);
- SendResult r = new SendResult();
- r.setMessageId(event.getId());
- r.setTopic(topic);
- cb.onSuccess(r);
- }
-
- @Override
- public List<EventMeshFrame> poll(String topic, int partition, long startOffset, int maxEvents, long timeoutMs) {
- Queue<CloudEvent> q = queues.get(topic);
- if (q == null) {
- return new ArrayList<>();
- }
- List<EventMeshFrame> out = new ArrayList<>();
- CloudEvent e;
- while (out.size() < maxEvents && (e = q.poll()) != null) {
- out.add(EventMeshFrame.fromCloudEvent(e));
- }
- return out;
- }
-
- @Override
- public void assignPartitions(String topic, List<Integer> partitions) {
- // no-op
- }
-
- @Override
- public void commitOffset(String topic, int partition, long offset) {
- // no-op
- }
-
- @Override
- public boolean isStarted() {
- return true;
- }
-
- @Override
- public boolean isClosed() {
- return false;
- }
-
- @Override
- public void start() {
- // no-op
- }
-
- @Override
- public void shutdown() {
- // no-op
- }
- }
-}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/NacosClusterForwardIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/NacosClusterForwardIntegrationTest.java
deleted file mode 100644
index 4c1b671..0000000
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/NacosClusterForwardIntegrationTest.java
+++ /dev/null
@@ -1,252 +0,0 @@
-/*
- * 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.it;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-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.common.wire.EventMeshFrame;
-import org.apache.eventmesh.runtime.admin.UniAdminService;
-import org.apache.eventmesh.runtime.cluster.ClusterCoordinator;
-import org.apache.eventmesh.runtime.cluster.ClusterMembership;
-import org.apache.eventmesh.runtime.cluster.ClusterSubscriptionStore;
-import org.apache.eventmesh.runtime.cluster.HttpForwarder;
-import org.apache.eventmesh.runtime.cluster.MetaStore;
-import org.apache.eventmesh.runtime.cluster.NacosMetaStore;
-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.push.BufferedEvent;
-import org.apache.eventmesh.runtime.subscription.DistributionMode;
-
-import java.net.URI;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Queue;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.TimeUnit;
-
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
-
-import io.cloudevents.CloudEvent;
-import io.cloudevents.core.builder.CloudEventBuilder;
-
-/**
- * Two-instance cluster test against a REAL Nacos Meta (§13.2.6): a subscriber on instance A and a
- * publisher on instance B, where A's storage is empty so delivery can ONLY happen via B forwarding
- * — which requires B to learn of A's subscription through the Meta watch. This is the test that
- * proves the {@link NacosMetaStore} prefix-watch fix (NamingService.subscribe for {@code /em/subs/}),
- * since the InMemoryMetaStore path was already covered by {@code ClusterForwardIntegrationTest}.
- *
- * <p><b>Gated by {@code -Dit.nacos}</b>. No broker needed — each instance has its own in-memory
- * storage (only B's holds the published event).</p>
- */
-@EnabledIfSystemProperty(named = "it.nacos", matches = ".+")
-class NacosClusterForwardIntegrationTest {
-
- private static final String TOPIC = "nacos-forward-" + System.nanoTime();
-
- private Instance instA;
- private Instance instB;
-
- @AfterEach
- void tearDown() {
- if (instA != null) {
- instA.close();
- }
- if (instB != null) {
- instB.close();
- }
- }
-
- @Test
- void subscriberOnA_receivesViaForwardFromB_overRealNacos() throws Exception {
- String nacos = System.getProperty("it.nacos");
- // Each instance gets its OWN NacosMetaStore (own naming client + subSnapshot), as in
- // production — sharing one would let A's register populate the shared snapshot and mask the
- // cross-instance watch path under test.
- instA = boot(nacos, "A");
- instB = boot(nacos, "B");
-
- // Subscriber c1 on A (ingress.subscribe → ClusterCoordinator.subscribe → Nacos /em/subs/).
- instA.ingress.subscribe(TOPIC, "c1", DistributionMode.BROADCAST, null);
- instA.membership.heartbeat();
- instB.membership.heartbeat();
-
- // Wait for B to discover A's instance (HttpForwarder needs addressOf(A) to forward). The
- // instance heartbeat is a single put; B's naming client sees it via an async push. Nacos
- // push latency varies under load, so allow generous time.
- long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30);
- while (!instB.membership.liveInstances().contains(instA.selfId) && System.nanoTime() < deadline) {
- Thread.sleep(200);
- instA.membership.heartbeat(); // refresh until B sees it (heartbeat lease is short)
- }
- assertTrue(instB.membership.liveInstances().contains(instA.selfId),
- "B should discover A's instance via Nacos NamingService");
-
- // Wait for B to learn of A's subscription via the Nacos NamingService watch (the fix under
- // test). Until B's subStore sees c1, B's dispatch would find no target and not forward.
- deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30);
- while (instB.subStore.targetsFor(TOPIC, null).isEmpty() && System.nanoTime() < deadline) {
- Thread.sleep(200);
- }
- assertTrue(!instB.subStore.targetsFor(TOPIC, null).isEmpty(),
- "B should discover A's subscription via the Nacos /em/subs/ watch");
-
- // Publish on B (lands in B's storage only — A's storage is empty). B's pullAndDispatch pulls
- // it, the coordinator sees c1 lives on A (not self) → HttpForwarder POST /internal/forward
- // → A.deliverLocal → c1's buffer. A's own pullAndDispatch would find nothing (empty storage),
- // so receiving the event PROVES it came via the cross-instance forward.
- CloudEvent event = CloudEventBuilder.v1()
- .withId("nf-1").withSource(URI.create("it")).withType("it.event").build();
- instB.ingress.publish(TOPIC, event).get(5, TimeUnit.SECONDS);
- instB.ingress.pullAndDispatch(TOPIC, 100, 0L);
-
- List<BufferedEvent> received = new ArrayList<>();
- deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
- while (received.isEmpty() && System.nanoTime() < deadline) {
- received.addAll(instA.ingress.poll("c1", 100, 100L));
- }
- assertEquals(1, received.size(), "A should receive the event via B's cross-instance forward");
- assertEquals("nf-1", received.get(0).getEvent().attributes().get("id"));
- }
-
- /** Boot one instance: own NacosMetaStore + own in-memory storage + traffic HTTP (for
- * /internal/forward) + cluster wired on that MetaStore. */
- private Instance boot(String nacos, String tag) throws Exception {
- MetaStore meta = new NacosMetaStore(nacos);
- MeshStoragePlugin storage = new InMemoryStorage();
- UniIngressService ingress = new UniIngressService(storage, new InMemoryOffsetStore());
- UniAdminService admin = new UniAdminService(ingress);
- UniHttpServer http = new UniHttpServer(ingress, admin);
- int port = http.start(0);
- String selfId = "localhost:" + port; // addressOf(selfInstanceId) must be a reachable host:port
-
- ClusterMembership membership = new ClusterMembership(meta, selfId, selfId, 15_000L, System::currentTimeMillis);
- HttpForwarder forwarder = new HttpForwarder(membership);
- ClusterSubscriptionStore subStore = new ClusterSubscriptionStore(meta);
- ClusterCoordinator coordinator = new ClusterCoordinator(selfId, subStore,
- (topic, clientId, event) -> {
- ingress.deliverLocal(topic, clientId, event);
- return true;
- }, forwarder);
- ingress.withCluster(coordinator);
- return new Instance(ingress, http, membership, subStore, port, selfId);
- }
-
- private static final class Instance {
-
- final UniIngressService ingress;
- final UniHttpServer http;
- final ClusterMembership membership;
- final ClusterSubscriptionStore subStore;
- final int port;
- final String selfId;
-
- Instance(UniIngressService ingress, UniHttpServer http, ClusterMembership membership,
- ClusterSubscriptionStore subStore, int port, String selfId) {
- this.ingress = ingress;
- this.http = http;
- this.membership = membership;
- this.subStore = subStore;
- this.port = port;
- this.selfId = selfId;
- }
-
- void close() {
- try {
- membership.leave();
- } catch (Exception ignored) {
- // best-effort
- }
- http.stop();
- }
- }
-
- // ---- per-instance in-memory storage (only B's holds the published event) ----
-
- static final class InMemoryStorage implements MeshStoragePlugin {
-
- private final ConcurrentHashMap<String, Queue<CloudEvent>> queues = new ConcurrentHashMap<>();
-
- @Override
- public void init(java.util.Properties p) {
- // no-op
- }
-
- @Override
- public void send(String topic, EventMeshFrame frame, SendCallback cb) {
- CloudEvent event = frame.toCloudEvent();
- queues.computeIfAbsent(topic, k -> new ConcurrentLinkedQueue<>()).offer(event);
- SendResult r = new SendResult();
- r.setMessageId(event.getId());
- r.setTopic(topic);
- cb.onSuccess(r);
- }
-
- @Override
- public List<EventMeshFrame> poll(String topic, int partition, long startOffset, int maxEvents, long timeoutMs) {
- Queue<CloudEvent> q = queues.get(topic);
- if (q == null) {
- return new ArrayList<>();
- }
- List<EventMeshFrame> out = new ArrayList<>();
- CloudEvent e;
- while (out.size() < maxEvents && (e = q.poll()) != null) {
- out.add(EventMeshFrame.fromCloudEvent(e));
- }
- return out;
- }
-
- @Override
- public void assignPartitions(String topic, List<Integer> partitions) {
- // no-op
- }
-
- @Override
- public void commitOffset(String topic, int partition, long offset) {
- // no-op
- }
-
- @Override
- public boolean isStarted() {
- return true;
- }
-
- @Override
- public boolean isClosed() {
- return false;
- }
-
- @Override
- public void start() {
- // no-op
- }
-
- @Override
- public void shutdown() {
- // no-op
- }
- }
-}
