feat(cluster): PARTITION_OWNED_PULL delivery topology (issue #5309, closes) - #5317
Merged
qqeasonchen merged 10 commits intoAug 31, 2026
Merged
Conversation
…5302 Sub-PR D1) Brings the A2A Gateway back into the Runtime, this time with the unified control-plane stores from apache#5301 Sub-PR A/C as the durable backend instead of the in-memory TaskRegistry that PR apache#5260 introduced. What lands ---------- * A2AGatewayService — task lifecycle controller. Replaces the in-memory TaskRegistry with TaskStore (Sub-PR A/C). A small runtime cache holds the two fields TaskStore does not model: parentTaskId (used to render child-task queries) and taskEpoch (the per-task counter the store requires for stale-write rejection on updateStatus). Both caches are rebuildable from the store on a fresh JVM; the cache keys are taskIds, so the cache size tracks active tasks only. * A2AGatewayHttpHandler — Netty HTTP handler for the REST + SSE API (/a2a/tasks, /a2a/tasks/{id}, /a2a/tasks/{id}/stream, /a2a/health). SSE registers a StatusSubscriber so intermediate PENDING -> RUNNING -> COMPLETED transitions stream to the client. * A2AGatewayServer — Netty HTTP bootstrap. Production wires EventMeshA2ATransport (Runtime-bridged) for the A2AMessageTransport dependency; tests pass an in-process transport. The weather-agent demo from PR apache#5260 is removed — the demo client (A2AGatewayDemo) and the Meta-ized AgentCard registry land in Sub-PR D2. * AgentCardRegistry + InMemoryAgentCardRegistry — minimal discovery surface (isAgentRegistered / registerCard / getCard). A Meta-backed implementation backed by SessionStore (Sub-PR A) lands in D2. Status mapping (PR apache#5260 -> Sub-PR A/C) --------------------------------------- SUBMITTED -> PENDING, WORKING -> RUNNING, CANCELLED -> CANCELED (one L) The legacy wire vocabulary stays available via A2AGatewayService.TaskState and the toLegacyState() helper. JSON response payloads still emit SUBMITTED/WORKING/CANCELLED so existing A2A clients do not break. Documentation ------------- * docs/a2a-protocol/README.md carries an EXPERIMENTAL banner that lists the three pieces still missing for production (TaskExpirer, Meta-ized AgentCard, Testcontainers E2E) — all scheduled for Sub-PR D2. Acceptance against issue apache#5302 ------------------------------ * "A2A transport over the Runtime delivery path" — A2AGatewayService takes an A2AMessageTransport; production wires EventMeshA2ATransport which already bridges A2A onto UniIngressService (issue apache#5301 Sub-PR A's runtime ingress). The parallel InMemoryA2AMessageTransport is no longer required by the gateway path. * "A2A tasks persist through TaskStore and recover across restarts" — createTask / updateStatus / getTask all go through the persistent store. On a fresh JVM, a gateway that had pending tasks sees them with their last persisted status; in-flight work that was awaiting a response will be picked up on the next status callback. Tests (7, all passing) ---------------------- * A2AGatewayServiceTest — 6 tests: create+complete, cancel, cancel-on- unknown, submit-to-unregistered, parent-child index, task-not-found. Uses an in-process TaskStore (mirrors Sub-PR A's test stub) and an in-process A2AMessageTransport that handles A2A's + single-segment wildcards. * A2AGatewaySmokeTest — Netty HTTP /a2a/health loopback check. Not in D1 (Sub-PR D2) --------------------- * TaskExpirer reaper (periodic TaskStore.expireStale sweep). * Meta-backed AgentCardRegistry (SessionStore). * Testcontainers fault-injection E2E.
…5313) A2AGatewayHttpHandler.java: - Remove unused imports (A2AProtocolConstants, A2ATopicFactory, TaskState) - Expand try-catch blocks for limit/offset parsing (EmptyCatchBlock, NeedBraces) - Expand handleList loop control flow (NeedBraces) - Collapse duplicate blank line between package and imports (EmptyLineSeparator) A2AGatewayServiceTest.java: - Expand single-line if-blocks in InProcessTaskStore.listByAgent - Expand if-block in InProcessTransport.matches - Expand if-blocks in tearDown - Move taskId declaration closer to its use (VariableDeclarationUsageDistance) A2AGatewaySmokeTest.java: - Rewrite StubTaskStore / NoopTransport with multi-line method bodies (LeftCurly, RightCurlyAlone, NeedBraces, EmptyLineSeparator) - Add missing static imports (assertEquals, assertNotNull) Build PR apache#5313 fix.
…owup) A2AGatewayServiceTest.java: - Fix ImportOrder: AgentCapabilities before AgentCard (alphabetical) - Fix VariableDeclarationUsageDistance for f: inline taskId into submitTask call so the declaration is closer to first use A2AGatewaySmokeTest.java: - Fix ImportOrder: static Assertions imports before non-static TaskStore import Build PR apache#5313 followup.
…ollowup) The previous fixup broke the test logic by calling TaskResult.getTaskId() which does not exist. Restore the original taskId declaration pattern while keeping the f declaration within 3 lines of its first use: - Declare taskId first - Declare f immediately after - Inline cancelTask result into the assertTrue (no intermediate boolean) - Move f.get() call right after the cancel assertion so f's first use is within 3 lines of its declaration Build PR apache#5313 followup.
…ache#5309) Sub-step 2 of apache#5309: PARTITION_OWNED_PULL delivery topology. Adds DeliveryTopology enum with two modes:/n- LOCAL_STICKY_PULL (default): every instance polls all partitions - PARTITION_OWNED_PULL: CAS+fencing ownership per partition (scale-out) fromConfig() parses eventmesh.delivery.topology property:/n- null/blank -> LOCAL_STICKY_PULL (backward compatible) - unknown -> IllegalArgumentException (fail-fast)
…ache#5309 sub-step 3) Boots PartitionOwnership state machine when topology=PARTITION_OWNED_PULL:/n- New 9-arg constructor with DeliveryTopology, instanceId, instanceAddress - start() calls startPartitionOwnership() after offset alignment - startPartitionOwnership(): ClusterMembership + PartitionOwnership + InMemoryMetaStore - shutdown() calls stopPartitionOwnership() before pull-loop shutdown - Backward-compatible 6-arg constructor defaults to LOCAL_STICKY_PULL
…ser tests (apache#5309 sub-step 4) UniRuntimeTopologyWiringTest boots real UniRuntime instances (full start()/shutdown() lifecycle, real scheduler) against a shared InMemoryMetaStore:/n- two PARTITION_OWNED_PULL instances converge on a disjoint, full-coverage split - LOCAL_STICKY_PULL default (6-arg ctor) never assigns partitions (poll-all) - null topology fails fast in the constructor DeliveryTopologyTest covers the fromConfig parser: null/blank -> default, exact names, whitespace trim, unknown/lowercase -> IllegalArgumentException. Complements ClusterDeliveryFaultTest (deterministic refreshOnce fault injection without a runtime). A full Testcontainers Kafka+Nacos E2E is tracked as a follow-up.
… (issue apache#5309 sub-step 5) Updates the §13.2 status banner (blocker closed) and adds §13.2.10 documenting the two legitimate topologies: - LOCAL_STICKY_PULL: default, single-instance, no Meta dependency (backward compatible) - PARTITION_OWNED_PULL: scale-out, requires Meta CAS + fencing (production HA) Includes selection semantics (fromConfig), startup flow (UniRuntime.start partition ownership), diff vs the §13.2.8 original design, and verification coverage (unit + E2E wiring + fault injection; Testcontainers deferred as a follow-up).
…13.2.10 with apache#5293 entry)
qqeasonchen
added a commit
to qqeasonchen/eventmesh
that referenced
this pull request
Aug 31, 2026
…pache#5314) Documents the 6-scenario CrossStoreFaultInjectionTest: why single-store contract tests cannot cover cross-store invariants, the four harness primitives, the per-scenario assertion table, why Testcontainers is not used, and the two extension points (RocksDB-backed crash scenario, MetaBackedOffsetStore going active). Numbered 13.2.12 because §13.2.11 was taken by the dual-topology matrix in PR apache#5317.
qqeasonchen
added a commit
to qqeasonchen/eventmesh
that referenced
this pull request
Aug 31, 2026
…pache#5314) Documents the 6-scenario CrossStoreFaultInjectionTest: why single-store contract tests cannot cover cross-store invariants, the four harness primitives, the per-scenario assertion table, why Testcontainers is not used, and the two extension points (RocksDB-backed crash scenario, MetaBackedOffsetStore going active). Numbered 13.2.12 because §13.2.11 was taken by the dual-topology matrix in PR apache#5317.
qqeasonchen
added a commit
that referenced
this pull request
Aug 31, 2026
…ntrol plane (issue #5314) (#5318) * test(state): add cross-store fault-injection harness (issue #5314) Four in-process primitives shared by the six #5314 scenarios: MetaPartitionSwitch - MetaStore wrapper; open()/close() simulates a network partition (mutating ops throw MetaPartitionException, reads continue against the pre-partition snapshot). CrossStoreRaceProbe - ordered log of cross-store operations (DELIVERY_PUT/ REMOVE, OFFSET_WRITE/READ, TASK_UPDATE) with a monotonic seq so a test can assert happens-before relationships. JvmCrashHarness - child JVM + sentinel-file-driven destroyForcibly() (SIGKILL / TerminateProcess), then relaunch against the same on-disk stores. Gated on ENABLE_JVM_CRASH_HARNESS. InMemorySubscriptionStore- ConcurrentHashMap-backed SubscriptionStore for the split-brain scenario (two views of the world). All four are test-only and run fully in-process: no Nacos, no Docker, no Testcontainers. See §13.2.12 of the architecture doc (added in a follow-up commit). * test(state): add cross-store fault-injection 6-scenario test (issue #5314) CrossStoreFaultInjectionTest covers the fault modes that span two or more stores (or the runtime + cluster-shared Meta). The individual store contract tests (#5310/#5311/#5312/#5313) cannot observe these, because the invariant lives at the seam: 1. CrashMidAckReAck - crash after offset-write, before MQ-ACK callback; recovery retires without re-invoking the channel (issue #5291 idempotency). 2. MetaPartitionDuringDlq - Meta unreachable while dead-letter recording; the store throws MetaPartitionException rather than silently no-op'ing, so the dispatcher keeps the delivery in flight and retries on heal (#5292). 3. A2aCancelMidStream - cancel lands between PENDING and RUNNING; the taskEpoch guard rejects stale-epoch late transitions and the task converges on one terminal state (#5302). 4. SubscriptionReRegisterAfterSplit - update during a Meta partition; after heal the latest write wins, nothing dropped or duplicated (#5288, #5301 SubscriptionStore). 5. OffsetStoreRaceVsDeliveryStore - cross-thread offset-advance vs retire race; the probe log proves every DELIVERY_REMOVE is preceded by an OFFSET_WRITE at the same offset (#5289 at-least-once). 6. A2aDispatchRaceVsTaskStore - two dispatchers race on one task record; stale-epoch writes are rejected and createTask yields exactly one winner (#5291). Every scenario runs in-process and deterministically. The JvmCrashHarness from the previous commit remains the optional cross-JVM verification path. Note on the taskEpoch contract exercised by scenarios 3 and 6: the epoch is set at createTask and never reset, so updateStatus rejects any epoch that differs from the record's. Same-epoch writes are last-writer-wins by design - the Runtime dispatcher is the sole writer and the epoch guards against a restarted instance's stale handle, not against intra-JVM ordering. * docs(architecture): add §13.2.12 cross-store fault-injection (issue #5314) Documents the 6-scenario CrossStoreFaultInjectionTest: why single-store contract tests cannot cover cross-store invariants, the four harness primitives, the per-scenario assertion table, why Testcontainers is not used, and the two extension points (RocksDB-backed crash scenario, MetaBackedOffsetStore going active). Numbered 13.2.12 because §13.2.11 was taken by the dual-topology matrix in PR #5317.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Overview
feat(cluster): PARTITION_OWNED_PULL delivery topology(issue #5309, closes).One-shot full PR covering all 5 sub-steps of #5309.
Changes (5 commits)
feat(cluster): add DeliveryTopology enum + fail-fast parserDeliveryTopologyenum (LOCAL_STICKY_PULL/PARTITION_OWNED_PULL)fromConfig(String)parser: null/empty → default; unknown value →IllegalArgumentException(fail-fast)feat(cluster): wire PARTITION_OWNED_PULL topology into UniRuntimeLOCAL_STICKY_PULL,"standalone"), new 9-arg addedstart()invokesstartPartitionOwnership()whentopology == PARTITION_OWNED_PULLstartPartitionOwnership()constructsClusterMembership+PartitionOwnership+InMemoryMetaStoreshutdown()callsstopPartitionOwnership()first to release Meta assignment recordstest(cluster): PARTITION_OWNED_PULL wiring E2E + parser testsDeliveryTopologyTest(7 cases): null/empty/blank/exact name/trim/unknown value/lowercaseUniRuntimeTopologyWiringTest(3 cases): two instances of PARTITION_OWNED_PULL converge to a full-coverage disjoint split; LOCAL_STICKY_PULL never assigns partitions; null topology fails fastdocs(uni-arch): add DeliveryTopology dual-topology matrix to §13.2.10fix(docs): rename dual-topology section to §13.2.11Dual-Topology Matrix Summary
ClusterMembership+PartitionOwnership+ heartbeatwithClusterMeta()injectionVerification
./gradlew :eventmesh-runtime:checkstyleMain :checkstyleTestpasses:eventmesh-runtime:test --tests DeliveryTopologyTest7/7 pass:eventmesh-runtime:test --tests UniRuntimeTopologyWiringTest3/3 pass:eventmesh-runtime:test --tests ClusterDeliveryFaultTest5/5 pass (no regression):eventmesh-runtime:test --tests UniRuntimeTest1/1 pass (no regression)Follow-ups
EventMeshApplicationintegration with theeventmesh.delivery.topologyconfig key: not included in this PRRelated
PartitionOwnership @Deprecatedis removed,MetaBackedOffsetStoreetc. can proceed)