apache · qqeasonchen · Aug 20, 2026 · Aug 19, 2026
diff --git a/docs/eventmesh-uni-architecture-redesign.md b/docs/eventmesh-uni-architecture-redesign.md
@@ -2872,6 +2872,79 @@ T4  对齐完成，标记 Meta=HEALTHY，退出降级
 
 > **降级哲学**：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
@@ -110,8 +110,7 @@ final List<String> BROKER_IT_5X_CLASSES = [
     '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
@@ -51,7 +51,7 @@ public class EventMeshApplication {
     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,30 +147,51 @@ public EventMeshApplication(MeshStoragePlugin storage, OffsetStore offsetStore,
     /** 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(() -> {
                 lm.sample();
                 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 @@ public void start() throws Exception {
             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 @@ public void shutdown() {
         }
         // §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
@@ -183,6 +183,14 @@ public UniIngressService ingress() {
         return ingress;
     }
 
+    /**
+     * 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

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
@@ -100,10 +100,11 @@ public int dispatch(String topic, org.apache.eventmesh.common.wire.EventMeshFram
     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 @@ private List<ClusterSub> selectByMode(List<ClusterSub> targets, org.apache.event
                     : 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
@@ -48,18 +48,22 @@ public class ClusterMembership {
     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
@@ -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);
+        }
+    }
+}