[Bug] Unify cluster delivery topology and fence stale partition owners

[qqeasonchen]

Problem

Cluster-related code currently retains three partially overlapping models:

1. sticky local pull/local delivery configured by EventMeshApplication.enableCluster();
2. partition-owned pull through PartitionOwnership;
3. cross-instance forwarding through ClusterCoordinator and HttpForwarder.

Their startup wiring, offset ownership, ACK semantics, failure behavior, and production maturity are not represented by one explicit topology choice. PartitionOwnership also relies on soft generation fencing and read/write Meta updates that are not a compare-and-set lease protocol.

Relevant code:

• eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshApplication.java
• eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/UniIngressService.java
• eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/PartitionOwnership.java
• eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/cluster/ClusterMembership.java

Reproduction

1. Configure two instances against a shared MetaStore and Kafka topic with multiple partitions.
2. Introduce a network partition or MetaStore outage for one instance.
3. Continue publishing and consuming during the split.
4. Restore connectivity and compare per-partition delivery sets, duplicates, offsets, and ownership records.
5. Repeat for sticky mode and partition-owned mode.

Expected behavior

The deployment must select one explicit delivery topology with documented guarantees. Ownership handoff must prevent stale owners from continuing to create valid deliveries after their epoch/lease is lost.

Proposed direction

Abstract the cluster mode into a single configurable DeliveryTopology mapped from the existing DistributionMode enum. There is no cross-instance forwarding: broadcast and multicast are realized by every instance pulling the same message from the MQ and delivering it to its locally matched subscribers. The legacy ClusterCoordinator / HttpForwarder forwarding model is deprecated, not retained as a topology.

DistributionMode	Delivery topology	Semantics
BROADCAST / MULTICAST	STICKY_LOCAL_PULL	every instance pulls the same message from the MQ, then delivers to locally matched subscribers
LOAD_BALANCE	PARTITION_OWNED_PULL	each queue / partition has exactly one Runtime owner

STICKY_LOCAL_PULL (current default, from BROADCAST / MULTICAST)

• Broadcast / multicast are realized as every instance pulling the same message from the MQ; each instance delivers it to its locally matched subscribers.
• No cross-instance forwarding.
• Supports global broadcast and filter-matched delivery.
• Trade-off: MQ pulls are amplified, because every Runtime pulls the full topic.

PARTITION_OWNED_PULL (high-throughput, from LOAD_BALANCE)

• Each queue / partition has exactly one Runtime owner.
• Requires owner fencing and handoff.
• No duplicate pulls and no cross-instance forwarding.
• Senders may optionally pass a partitionKey so that the same key is routed to a stable partition, preserving per-key ordering.

DistributionMode cleanup

• Remove LOAD_BALANCE_STICKY from DistributionMode.java. It is no longer needed: the per-key ordering it reserved is now covered by PARTITION_OWNED_PULL's optional partitionKey.

Each topology must document: duplicate-delivery semantics, cost / throughput characteristics, failure-recovery strategy, and a corresponding E2E test.

• Define subscription persistence, physical cursor ownership, ACK target, duplicate bound, and Meta outage behavior for each topology.
• For partition ownership, use lease revision/CAS plus owner epoch fencing; carry the epoch into polling/delivery validation.
• Keep unfinished topologies experimental until their failover contract is tested.

Acceptance criteria

• Runtime startup selects one topology explicitly; inactive topology components are not silently half-wired.
• DistributionMode maps unambiguously to STICKY_LOCAL_PULL or PARTITION_OWNED_PULL; neither topology uses cross-instance forwarding.
• LOAD_BALANCE_STICKY is removed from DistributionMode.java.
• Documentation defines ACK, offset, duplicate, and recovery semantics per topology.
• Partition handoff uses an atomic lease/revision or equivalent fencing mechanism.
• A stale owner cannot continue valid delivery after losing ownership.
• Fault-injection tests cover Meta outage, network split, owner crash, graceful handoff, and recovery.
• Kafka multi-instance tests verify no unintended simultaneous partition ownership.

Suggested labels

bug, need research, architecture, cluster, priority:p1

---

Supersedes #5300 (Configurable DeliveryTopology for cluster delivery), which is folded into this issue's two-topology definition.

[qqeasonchen]

The proposed direction has been implemented and submitted as #5308 (single PR, based on current develop):

• Sticky-only topology: HttpForwarder and the /internal/forward + /internal/reply-forward endpoints are deleted; reply() on a non-owner instance now 404s (the SDK pins to one instance via instanceUrl, so cross-instance forwarding is no longer needed). DistributionMode.LOAD_BALANCE_STICKY is removed; LOAD_BALANCE absorbs the sticky semantics (partition-key hash routes to one subscriber).
• Meta CAS fencing: MetaStore.tryAcquire(key, expectedOldValue, newValue) atomic CAS (InMemory ConcurrentHashMap / Nacos 2.x publishConfigCas with casMd5); FencingToken "<bootEpoch>:<counter>" per JVM, monotonic and restart-safe; assignment record /em/assignments/<topic#partition> = "<token>|<ownerInstanceId>"; acquireOrFence 3-branch protocol with TTL-eviction takeover; releaseStale CAS'es a "" tombstone for partitions that left our assigner share so the new rightful owner can claim them (prevents stranding after membership churn).
• Fault-injection tests (ClusterDeliveryFaultTest): in-process 3-4 instance scenarios (steady-state split, crash takeover, scale-out churn, Meta partition split-brain guard, healed partition reclaim) driven by a mutable clock — fully deterministic, no sleeps.

Design details: docs/eventmesh-uni-architecture-redesign.md §13.2.10.