EventMesh control plane

Audience: operators and contributors. What the runtime boot path wires from configuration, and how every state store fails when its backend goes down.

---

This page documents the control-plane components that the runtime boot path wires from configuration. For each component, we list the config key, the default value, the failure behavior when the key is missing or malformed, and the ingress route(s) that the component gates.

1. DeliveryTopology (#5293, #5309)

Selects how the runtime polls / dispatches events. Two modes:

Mode	Default?	What it does	What it requires
LOCAL_STICKY_PULL	Yes	Every runtime instance polls every partition of every subscribed topic. No Meta store dependency. Backward compatible with pre-#5301 deployments.	None
PARTITION_OWNED_PULL	No	Each instance acquires ownership of a strict subset of partitions via Meta CAS + fencing. No duplicate consumption across instances.	Meta store (Nacos today). Code is intact (see PartitionOwnership); the wiring follow-up is tracked in #5309.

1.1 Configuration

• Key: eventmesh.delivery.topology
• Default: LOCAL_STICKY_PULL (the value is read in EventMeshApplication via DeliveryTopology.fromConfig(System.getProperty(...))).
• Set via: JVM -D flag (-Deventmesh.delivery.topology=PARTITION_OWNED_PULL) or the EVENTMESH_DELIVERY_TOPOLOGY environment variable (the runtime forwards env-var values into system properties at boot).
• Documented in eventmesh.properties: the property is listed with a commented default and an inline explanation of both modes; operators can uncomment + edit without leaving the config file.

1.2 Failure behavior

• Missing or blank -> resolved to LOCAL_STICKY_PULL (backward compatible). The startup log emits delivery topology=LOCAL_STICKY_PULL.
• Unknown value (e.g. PARTITION_OWNER typo) -> IllegalArgumentException raised by DeliveryTopology.fromConfig. The runtime fails fast at boot; the operator sees a clear error in bin/start.sh output. A typo must not silently degrade to single-instance mode (see DeliveryTopologyTest).
• PARTITION_OWNED_PULL without a Meta store -> enableCluster is required. The boot path throws if metaStore is null when the topology is PARTITION_OWNED_PULL.

1.3 Coverage per ingress route

The delivery topology is referenced from a single code path (UniRuntime.poll -> ownedPartitions(topic)), which is the consumer of UniIngressService.deliver. Therefore every ingress path is covered by construction:

Ingress route	Path through topology
HTTP POST /events/publish	UniHttpServer -> UniIngressService.publish -> Frame -> Producer.send -> poll loop -> topology-selected partition set
A2A POST /a2a/tasks/send	A2AGatewayHttpHandler -> A2AGatewayService -> UniIngressService -> same as above
Legacy TCP (UniTcpServer)	main path is being phased out (see docs/feature/protocols.md #5341); when active, the TCP server feeds the same UniIngressService
WebSocket push	passive receiver; not gated by topology (the topology decides which partitions to poll, not the push path)
SSE	passive receiver; not gated by topology

Coverage is verified by git grep of DeliveryTopology in the runtime source tree (only UniRuntime and DeliveryTopology itself should match in eventmesh-runtime/src/main/).

2. SecurityGate (#5304)

SecurityGate is the per-request control-plane gate that lives in front of UniIngressService and A2AGatewayService. It is composed of:

• FilterChain (auth / acl): TokenAuthFilter, SignatureVerifierFilter, AclFilter.
• QuotaManager (per-tenant): TenantQuotaManager.
• AuditSink: pluggable (default is LoggingAuditSink).

2.1 Configuration

• Key: eventmesh.security.gate.* (see eventmesh.properties section eventmesh.security.gate.* and the per-filter keys).
• Default: a no-op gate that allows every request. This is the backward-compatible default and is documented as such in docs/quickstart/configuration.md.
• Fail-closed mode: when eventmesh.security.gate.failClosed=true, any filter chain exception (auth failed, signature invalid, ACL denied) results in a 401/403 response. The default is fail-open (allowed) so a misconfigured gate does not silently break a deployment.

2.2 Failure behavior

• Missing or blank -> no-op gate; every request is allowed. Operators who need a real gate must set the per-filter keys.
• Malformed ACL entry -> logged at WARN, the request is denied (fail safe in the security context), the deny is counted in the gate's metrics counter.
• Audit sink failure -> logged at ERROR, the request is still allowed (audit is best-effort; a failed audit must not take down the data path).

2.3 Coverage per ingress route

Ingress route	Gated by SecurityGate?	Where the gate is invoked
HTTP POST /events/publish	Yes	UniHttpServer -> RequestContext -> SecurityGate.check(ctx, frame) -> UniIngressService.publish
HTTP POST /events/subscribe	Yes	same chain
A2A POST /a2a/tasks/send	Yes	A2AGatewayHttpHandler -> gate -> A2AGatewayService -> UniIngressService
Legacy TCP	Yes (gated by TCP-level auth, separate from SecurityGate)	UniTcpServer -> TcpIngressBridge (the TCP path has its own connection-level auth, see docs/feature/protocols.md)
Admin HTTP	Yes	UniAdminServer -> UniAdminService (separate gate, configured under eventmesh.admin.security.*)
WebSocket push	n/a (passive)	push does not invoke the gate; the gate ran at subscribe time
SSE	n/a (passive)	same as WS
Connector Runtime	Yes	ConnectorScheduler -> gate (configured separately)

3. Configuration matrix

Property	Default	Component	Effect of the default
eventmesh.delivery.topology	LOCAL_STICKY_PULL	DeliveryTopology	Single-instance / no Meta dependency
eventmesh.security.gate.failClosed	false	SecurityGate	Backward-compatible allow-all
eventmesh.storage.type	(commented out)	Storage plugin	Must be set to a supported backend
eventmesh.connector.offset.store	memory	Connector	Offsets not persisted (dev-only)

4. References

• Parent: #5296 (Architecture Review, Q2 / 2026-09-07)
• Tracking: #5338
• DeliveryTopology decision: #5293 (closed via PR #5308, implementing LOCAL_STICKY_PULL)
• DeliveryTopology wiring follow-up: #5309 (PARTITION_OWNED_PULL wiring)
• SecurityGate: #5304
• Architecture: docs/architecture/overview.md (section 4, Security gate)
• Configuration: docs/quickstart/configuration.md

State store failure matrix

Absorbed from the per-backend failure analysis (issue #5339). What happens when each store's backend goes down, and how the restart / multi-instance / fencing scenarios are covered by StateStoreDurabilityTest.

1. Store-to-backend map

Store	Backend	Tier	Production impl	Test impl
OffsetStore	RocksDB (local)	L1	RocksDBOffsetStore	InMemoryOffsetStore (degraded mode only)
SubscriptionStore	Meta + local cache	L2	ClusterSubscriptionStore (watches /em/subs/)	state.fault.InMemorySubscriptionStore
SessionStore	Meta + local cache	L2	SessionRegistry (agents, bindings, sessions)	direct SessionRegistry over InMemoryMetaStore
DeliveryStateStore	RocksDB (local, sub-second flush)	L1	RocksDBDeliveryStateStore	InMemoryDeliveryStateStore
DeadLetterStore	Meta CAS	L3	MetaBackedDeadLetterStore (/em/dlq/)	direct wrapper over InMemoryMetaStore
TaskStore	Meta CAS	L3	MetaBackedTaskStore (/em/tasks/)	direct wrapper over InMemoryMetaStore

2. Per-backend failure behavior

2.1 RocksDB backend (OffsetStore, DeliveryStateStore)

Scenario	Behavior	Operator-visible signal	Recovery
Local disk full	writeOffset / put returns false or throws IllegalStateException	RocksDBOffsetStore.offsetWriteFailures counter; ReliableDispatcher.pendingCount does not drop	Operator frees disk; next write succeeds. The dispatcher keeps the delivery in flight (issue #5290).
Process kill -9 (no graceful close)	The RocksDB file is the durability surface; the next process open at the same path sees the last flushed state	n/a (no log emitted on the way down)	The runtime calls UniRuntime.alignPullOffsetsToAck + ReliableDispatcher.recover() on boot. StateStoreDurabilityTest#DeliveryStateStoreKillMinusNine exercises this path.
Process restart (graceful)	flush() is called by the shutdown hook, then close(). Re-open at the same path is a no-op for the persisted entries	n/a	Same as kill -9.
Corrupted RocksDB file	RocksDB.open throws RocksDBException -> IllegalStateException at the constructor. The runtime fails fast at boot.	IllegalStateException("failed to open RocksDB offset store at <path>") in the boot log.	Operator restores from backup or wipes the data directory (data loss is unavoidable in this case; the runtime fails fast rather than silently losing state).

2.2 Meta backend (SubscriptionStore, SessionStore, DeadLetterStore, TaskStore)

All four Meta-backed stores use the same primitives:

• MetaStore.putIfAbsent (CAS) for first-write-wins (DLQ, TaskStore.createTask).
• MetaStore.tryAcquire(expectedOldValue, newValue) (CAS) for status updates (TaskStore.updateStatus).
• MetaStore.put (last-write-wins) for plain updates (SubscriptionStore.put).
• MetaStore.get / getWithPrefix for reads.

Scenario	Behavior	Operator-visible signal	Recovery
Meta unreachable (network partition)	put / putIfAbsent / tryAcquire throw RuntimeException (the production NacosMetaStore wraps the Nacos exception)	The exception is caught at the call site; DeliveryDispatcher keeps the delivery in flight; A2AGatewayService surfaces the error to the A2A caller; ReliableDispatcher keeps the in-flight delivery in the ledger	Once Meta heals, the next tick succeeds. Idempotent on the retry: DeadLetterStore.recordDeadLetter is idempotent on already-present keys; TaskStore.createTask returns null for a duplicate taskId (the gateway generates a fresh id and retries).
Meta cluster split-brain (two instances write the same key)	First-write-wins (putIfAbsent) is correct; tryAcquire rejects the stale writer (returns false); plain put is last-write-wins	The losing writer observes false from the CAS call and re-reads	Operator doesn't need to act; the contract is self-healing. StateStoreDurabilityTest#TaskStoreMetaFailure exercises the partitioned-write case.
Meta returns stale data (clock skew)	MetaStore.tryAcquire(expectedOldValue=...) rejects writes whose expectedOldValue does not match the current Meta value; the caller re-reads	The call returns false	Caller re-reads via meta.get(key) and retries with the fresh value. The Meta-backed stores do not carry their own wall-clock dependency; staleness surfaces as a CAS mismatch, not as a time-based inconsistency.
Meta restarts (process restart of the Meta server)	All reads return the snapshot from the most recent successful write; writes resume once the new Meta is reachable	The production NacosMetaStore reconnects via the Nacos client retry loop; the runtime logs the reconnection at INFO	No operator action needed; the in-process caches (SessionRegistry.agentCache, ClusterSubscriptionStore.cache) are rebuilt from the next watch event. StateStoreDurabilityTest#DeadLetterStoreRestart exercises the wrapper-restart-over-shared-Meta case.
Local in-process cache diverges from Meta	The next read through MetaStore.get / getWithPrefix returns the Meta view; the in-process cache is refreshed on the next watch event	n/a (no log)	Convergence is automatic; the local cache is best-effort.

3. Restart / multi-instance / fencing coverage

Each row maps an issue #5339 acceptance scenario to the test that exercises it.

#	Scenario	Store	Test
1	Restart: persist offset 100, kill runtime, restart, read offset = 100	OffsetStore (RocksDB)	StateStoreDurabilityTest#OffsetStoreRestart.offsetSurvivesProcessRestart
2	Multi-instance: two ClusterSubscriptionStore instances sharing the same Meta; concurrent register / unregister converges	SubscriptionStore	StateStoreDurabilityTest#SubscriptionStoreMultiInstance.twoInstancesConvergeOnSameView (100-iteration storm, 8 threads) + removeIsObservedByAllInstances
3	Fencing: stale partition owner (instance A) writes after instance B takes over; the write is rejected	SessionStore	StateStoreDurabilityTest#SessionStoreFencing.staleHeartbeatAfterTakeoverIsRejected
4	Restart: DLQ ledger survives a wrapper restart (Meta is the durability surface)	DeadLetterStore (Meta)	StateStoreDurabilityTest#DeadLetterStoreRestart.ledgerSurvivesStoreWrapperRestart + writeSucceedsAfterMetaHeals
5	Cross-store failure: Meta goes down mid-submit; the TaskStore surfaces the failure rather than silently dropping the task	TaskStore (Meta)	StateStoreDurabilityTest#TaskStoreMetaFailure.createTaskFailsWhenMetaIsPartitioned + updateStatusFailsWhenMetaIsPartitioned
6	Kill -9: in-flight ledger survives abrupt close	DeliveryStateStore (RocksDB)	StateStoreDurabilityTest#DeliveryStateStoreKillMinusNine.inFlightLedgerSurvivesAbruptClose

4. Acceptance check (for #5339)

• For each of the 6 stores above, a JUnit test exists in eventmesh-runtime (StateStoreDurabilityTest).
• The test for each scenario above is referenced in the table in section 3.
• No production-code store is backed by an InMemory* implementation; the in-memory implementations live in state/fault/ (test-only).
• This document exists (docs/feature/control-plane.md).

5. References

• Parent: #5296 (Architecture Review, Q3 / 2026-09-07)
• Tracking: #5339
• Tests: eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/state/StateStoreDurabilityTest.java
• Related: #5289 (monotonic offset), #5290 (write failure semantics), #5291 (idempotency), #5292 (DLQ durability), #5301 (state control plane)
• Sub-PRs:
  • #5301 Sub-PR A (interfaces + SPI)
  • #5301 Sub-PR B (RocksDB-backed delivery + offset stores)
  • #5301 Sub-PR C (Meta-backed DLQ + TaskStore)
  • #5301 Sub-PR D (A2A Gateway on Runtime)