[Architecture Review][P0] PARTITION_OWNED_PULL delivery topology (reopens #5300 second mode)

[qqeasonchen]

Problem

Issue #5300 (closed as duplicate of #5293) originally proposed a three-mode DeliveryTopology:

• LOCAL_STICKY_PULL — current default after PR fix(cluster): unify delivery topology with Meta CAS + fencing (#5293 #5288) #5308; SDK pins via instanceUrl, single pull per partition, no forwarding
• PARTITION_OWNED_PULL — high-throughput mode with explicit partition ownership and fenced handoff
• FORWARDED_DISPATCH — centralized pull + cross-instance forwarding (compatibility shim)

PR #5308 (merged as 7260581 in develop) fully implemented the first mode. The PARTITION_OWNED_PULL code path (PartitionOwnership, ClusterCoordinator, MetaBackedOffsetStore) was deliberately left un-wired — enableCluster() carries a comment that says these classes are "intentionally NOT wired". They are fully unit-tested and the fencing protocol is correct, but no production code path activates them.

FORWARDED_DISPATCH was dropped on the rationale that the SDK's instanceUrl pin makes forwarding unnecessary.

#5300 was therefore closed prematurely: the second and third modes were not actually implemented anywhere, and no follow-up issue tracked them. This issue re-opens the tracking for PARTITION_OWNED_PULL only — FORWARDED_DISPATCH is intentionally out of scope.

Why this matters

Sticky local delivery has a known throughput ceiling. Once a partition is pinned to a subscriber via instanceUrl, the message rate that partition can sustain is bounded by that single instance's outbound HTTP / WebHook capacity. For high-fanout topics or partitions with skewed key distribution, this is the bottleneck.

PARTITION_OWNED_PULL is the natural next step:

• each EventMesh instance owns a strict subset of topic#partition assignments (Meta CAS via PartitionOwnership)
• the partition owner pulls from the MQ on behalf of its assigned partitions
• dispatch to the local subscribers is identical to the sticky path
• no cross-instance forwarding — the partition is owned, not just pinned to a routing point

This is the high-throughput mode EventMesh will need once it is deployed for sustained > 10K msg/s/part workloads.

Risks of leaving the code un-wired

The current enableCluster() comment ("intentionally NOT wired") is a maintenance hazard:

1. New contributors may call PartitionOwnership directly thinking it is production code, and end up writing assignments to /em/assignments/... in Meta that no other instance observes as authoritative.
2. The Meta namespace pollution is invisible until a future wire-up lands and finds stale keys from a half-wired experiment.
3. The class lacks @Deprecated(forRemoval=true) even though it is not on any active code path, so static analyzers do not flag it.
4. The 5 fault-injection tests in ClusterDeliveryFaultTest exercise the path, but production behavior under the same scenarios is unverified (it is the tests that bind the behavior, not the runtime).

Proposed direction

Land the PARTITION_OWNED_PULL topology as a second runtime mode. Concrete plan, in execution order:

Sub-step 1: Mark the un-wired code correctly

In eventmesh-runtime's cluster/ package:

• Add @Deprecated(forRemoval=true) to PartitionOwnership, ClusterCoordinator, ClusterMembership (the cluster-coupled bits, not the agent/session ClusterMembership).
• Add a Javadoc note on each: "Reserved for the PARTITION_OWNED_PULL delivery topology (issue TBD). Currently inactive in LOCAL_STICKY_PULL mode; not safe to call from production code."
• Keep the in-memory fault-injection tests (they are the contract this issue will wire against).

Sub-step 2: Introduce DeliveryTopology config

Add a new top-level config eventmesh.runtime.deliveryTopology with two legal values:

• LOCAL_STICKY_PULL (default, current behavior)
• PARTITION_OWNED_PULL (this issue)

Wire enableCluster() to consult this config and only activate PartitionOwnership when the second mode is selected. Misconfiguration (null, typo) must fail fast at boot, not silently fall back to sticky.

Sub-step 3: Wire the existing PartitionOwnership path

The classes already implement the protocol; the only missing piece is the activation. Specifically:

• enableCluster() calls partitionOwnership.start() when PARTITION_OWNED_PULL is selected
• UniIngressService.pullAndDispatch consults partitionOwnership.ownedPartitions(topic) to decide which partitions to pull (the existing single-instance fall-through to poll-all stays as the LOCAL_STICKY_PULL behavior)
• SubscriptionManager continues to deliver locally as today; no HttpForwarder is re-introduced

Sub-step 4: E2E tests for PARTITION_OWNED_PULL

Port the 5 in-process fault-injection scenarios in ClusterDeliveryFaultTest to a Testcontainers-backed integration test (Kafka + Nacos + 3 EM instances). The scenarios are:

• steady-state deterministic split (3 instances, 6 partitions, each instance owns 2)
• crashed-instance partitions are taken over by survivors after TTL
• membership churn (4th instance joins): only the re-assigned partitions move
• Meta partition: instance pauses polling until TTL expiry
• healed Meta partition: instance reclaims its share without spurious failover

This is the gap the original #5300 acceptance criterion called out ("E2E tests exist per topology (Meta outage, network split, owner crash)").

Sub-step 5: Documentation

Update docs/eventmesh-uni-architecture-redesign.md §13.2 with the matrix #5300 originally requested:

Topology	Duplicate semantics	Cost / throughput	Recovery
LOCAL_STICKY_PULL	zero (one pull, one delivery)	bounded by single instance outbound	per-subscriber offset replay on restart
PARTITION_OWNED_PULL	zero (owner is exclusive)	horizontal, scales with instance count	Meta CAS fencing, owner takeover on TTL expiry

Acceptance criteria

• eventmesh.runtime.deliveryTopology config selects the mode; boot fails fast on misconfiguration
• In PARTITION_OWNED_PULL mode, enableCluster() wires PartitionOwnership and the runtime pulls only owned partitions
• The un-wired code in cluster/ carries @Deprecated(forRemoval=true) with the pointer to this issue
• E2E integration tests cover the 5 fault scenarios for PARTITION_OWNED_PULL
• eventmesh-uni-architecture-redesign.md §13.2 documents both topologies with duplicate / cost / recovery semantics
• FORWARDED_DISPATCH remains out of scope (separate issue if ever needed)

Dependencies

• Phase 1 + 3 ([Bug] Make RocksDB and Meta offset writes atomically monotonic #5289 [Bug] Do not retire deliveries when durable offset persistence fails #5290 [Bug] Prevent stale ACKs from matching reused delivery IDs #5291 [Bug] Make DLQ transition durable before retiring source delivery #5292 [Bug] Unify cluster delivery topology and fence stale partition owners #5293 [Bug] Start and manage membership heartbeats in sticky cluster mode #5288) — DONE (PR fix(reliability): atomic offset writes, durable DLQ, epoch delivery IDs (#5289 #5290 #5291 #5292) #5307 + fix(cluster): unify delivery topology with Meta CAS + fencing (#5293 #5288) #5308 merged)
• Phase 2 [Architecture Review][P0] Unified state control-plane stores #5301 (Unified state control-plane stores) — strongly recommended before this lands, because PARTITION_OWNED_PULL mode increases the value of DeliveryStateStore persistence (cross-instance failover depends on durable state). Hard requirement only if we want to actually support the failover path with at-least-once guarantees; otherwise at-most-once is acceptable for the initial cut.
• eventmesh-meta (Nacos 2.x CAS) — already shipped

Related

• [Architecture Review][P0] Configurable DeliveryTopology for cluster delivery #5300 — original (now closed as duplicate) which proposed the three-mode matrix
• [Bug] Unify cluster delivery topology and fence stale partition owners #5293 — sticky mode implementation (PR fix(cluster): unify delivery topology with Meta CAS + fencing (#5293 #5288) #5308 ✓)
• [Architecture Review] Consolidate EventMesh boundaries and production-grade consistency #5296 — Architecture Review plan
• [Architecture Review][P0] Unified state control-plane stores #5301 — Unified state control-plane stores (Phase 2; pairs with this for at-least-once)
• eventmesh-uni-architecture-redesign.md §13.2 — to be updated in Sub-step 5