You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by ch...@apache.org on 2026/09/08 12:01:28 UTC
(eventmesh) branch develop updated: fix(#5359): wire DeliveryTopology + MetaStore into runtime boot; rollback on failure (#5370)
This is an automated email from the ASF dual-hosted git repository.
qqeasonchen pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/develop by this push:
new b3f9f75b4 fix(#5359): wire DeliveryTopology + MetaStore into runtime boot; rollback on failure (#5370)
b3f9f75b4 is described below
commit b3f9f75b40ffce2683740601bc0af0adf93aecf3
Author: Eason Chen <qq...@gmail.com>
AuthorDate: Tue Sep 8 20:01:22 2026 +0800
fix(#5359): wire DeliveryTopology + MetaStore into runtime boot; rollback on failure (#5370)
Phase 1 wiring of the production-HA plan (#5354), sub-issue #5359.
The gap: EventMeshApplication.enableCluster built a PartitionOwnership
state machine (Meta CAS + fencing) but never told the RUNTIME - the
pull loop kept its constructor-time topology (LOCAL_STICKY_PULL by
default) and polled every partition on every instance while the app
layer believed ownership was exclusive. Two divergent ownership views;
the cluster looked healthy but delivered duplicates.
Boot wiring:
- enableCluster now flips the runtime topology to PARTITION_OWNED_PULL
and injects the shared MetaStore BEFORE runtime.start(), so the pull
loop consults the same PartitionOwnership (ownedPartitions) the app
built - one state machine, not two
- UniRuntime.topology becomes volatile (flippable pre-start); new
withTopology / withClusterMeta builders reject post-start mutation
(fail fast, like the #5356 contract); new topology() / clusterMeta()
accessors
Startup transactionality (#5359):
- EventMeshApplication.start() wraps the whole boot in try/catch: on
any failure it runs shutdown() (stop schedulers, release ports,
release storage) with cleanup failures attached as suppressed, then
rethrows - no more half-started processes holding the admin port
- single-line effective-config log at startup (topology / metaWired /
wsPort / tls) so operators can audit a running deployment
Tests (new EventMeshApplicationStartupTest, 4 cases):
- enableCluster contract: topology flipped + shared meta injected
- withTopology / withClusterMeta fail fast after start
- null topology rejected
Local: runtime tests + checkstyle (maxWarnings=0) all green.
Refs #5359 (sub-issue of #5354).
---
.../runtime/boot/EventMeshApplication.java | 32 ++++-
.../apache/eventmesh/runtime/boot/UniRuntime.java | 38 ++++-
.../boot/EventMeshApplicationStartupTest.java | 159 +++++++++++++++++++++
3 files changed, 227 insertions(+), 2 deletions(-)
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
index e4925aa7d..abfc27dde 100644
--- 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
@@ -196,12 +196,42 @@ public class EventMeshApplication {
// 4. Dynamic config hot-reload.
new org.apache.eventmesh.runtime.cluster.DynamicConfigWatcher(metaStore, runtime.ingress()).start();
+ // 5. #5359: flip the runtime's pull topology to PARTITION_OWNED_PULL and inject the shared
+ // MetaStore BEFORE runtime.start() runs, so the pull loop consults this PartitionOwnership
+ // (ownedPartitions) instead of polling every partition on every instance (duplicate
+ // consumption). UniRuntime.startPartitionOwnership then reuses the injected meta instead of
+ // building a second, divergent ownership state machine.
+ runtime.withClusterMeta(metaStore);
+ runtime.withTopology(org.apache.eventmesh.runtime.cluster.DeliveryTopology.PARTITION_OWNED_PULL);
+
log.info("cluster enabled (sticky + partition fencing): instance={} token={}",
selfInstanceId, selfToken);
}
- /** Start runtime + traffic HTTP + admin HTTP. */
+ /** Start runtime + traffic HTTP + admin HTTP. Rolls back partial startup on failure (#5359). */
public void start() throws Exception {
+ try {
+ startupInternal();
+ } catch (Exception | Error failure) {
+ // #5359: a half-started process (runtime scheduler up, HTTP down) is worse than a
+ // clean exit - release ports, stop schedulers and rethrow so bin/start.sh surfaces it.
+ log.error("startup failed, rolling back: {}", failure.toString(), failure);
+ try {
+ shutdown();
+ } catch (RuntimeException cleanupFailure) {
+ failure.addSuppressed(cleanupFailure);
+ }
+ throw failure;
+ }
+ }
+
+ private void startupInternal() throws Exception {
+ // #5359: single-line effective configuration so an operator can audit a running
+ // deployment from the log (which topology / meta / storage actually took effect).
+ log.info("startup effective config: topology={} metaWired={} wsPort={} tls={}",
+ runtime.topology(), runtime.clusterMeta() != null
+ ? runtime.clusterMeta().getClass().getSimpleName() : "none",
+ wsPort, sslContext != null);
runtime.start();
runtime.ingress().registerRuntimeGauges();
UniAdminService adminService = new UniAdminService(runtime.ingress());
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
index 445c34789..490448a89 100644
--- 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
@@ -57,7 +57,8 @@ public class UniRuntime {
private final long pollTimeoutMs;
/** Delivery topology (§13.2). LOCAL_STICKY_PULL = poll-all (default); PARTITION_OWNED_PULL = CAS+fencing scale-out. */
- private final DeliveryTopology topology;
+ /** Set once at construction; may be flipped pre-start by EventMeshApplication.enableCluster (#5359). */
+ private volatile DeliveryTopology topology;
/** Instance identity for cluster membership (null in LOCAL_STICKY_PULL mode). */
private final String instanceId;
@@ -82,6 +83,41 @@ public class UniRuntime {
private Properties storageConfig = new Properties();
/** Inject storage config before {@link #start()}; additive, chainable. */
+ /** The delivery topology in effect (readable for startup logging, #5359). */
+ public DeliveryTopology topology() {
+ return topology;
+ }
+
+ /** The shared MetaStore wired by EventMeshApplication (null in single-instance mode, #5359). */
+ public MetaStore clusterMeta() {
+ return clusterMeta;
+ }
+
+ /** Wire the shared MetaStore before {@link #start()} (idempotent; used by enableCluster). */
+ public UniRuntime withClusterMeta(MetaStore metaStore) {
+ if (running.get()) {
+ throw new IllegalStateException("clusterMeta cannot change after start()");
+ }
+ this.clusterMeta = metaStore;
+ return this;
+ }
+
+ /**
+ * Flip the delivery topology pre-start (#5359). Only {@link EventMeshApplication#enableCluster}
+ * uses this: cluster mode upgrades a LOCAL_STICKY_PULL runtime to PARTITION_OWNED_PULL after the
+ * shared MetaStore is wired in. Must not be called after {@link #start()}.
+ */
+ public UniRuntime withTopology(DeliveryTopology newTopology) {
+ if (running.get()) {
+ throw new IllegalStateException("topology cannot change after start()");
+ }
+ if (newTopology == null) {
+ throw new IllegalArgumentException("topology must not be null");
+ }
+ this.topology = newTopology;
+ return this;
+ }
+
public UniRuntime withStorageConfig(Properties storageConfig) {
this.storageConfig = storageConfig;
return this;
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/boot/EventMeshApplicationStartupTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/boot/EventMeshApplicationStartupTest.java
new file mode 100644
index 000000000..c699098e1
--- /dev/null
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/boot/EventMeshApplicationStartupTest.java
@@ -0,0 +1,159 @@
+/*
+ * 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.boot;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.storage.MeshStoragePlugin;
+import org.apache.eventmesh.common.wire.EventMeshFrame;
+import org.apache.eventmesh.runtime.cluster.DeliveryTopology;
+import org.apache.eventmesh.runtime.cluster.InMemoryMetaStore;
+import org.apache.eventmesh.runtime.offset.InMemoryOffsetStore;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * Issue #5359 acceptance: boot-time wiring of DeliveryTopology + MetaStore + FencingToken.
+ *
+ * <p>Covers the three wiring contracts the plan demands: (a) enableCluster flips the runtime
+ * topology to PARTITION_OWNED_PULL and injects the shared MetaStore before start; (b) the
+ * pre-start builders reject post-start mutation; (c) PARTITION_OWNED_PULL without a MetaStore
+ * fails fast (the #5356 contract, re-verified at the application level).</p>
+ */
+class EventMeshApplicationStartupTest {
+
+ private static final class StubStorage implements MeshStoragePlugin {
+
+ @Override
+ public void init(Properties props) {
+ // no-op
+ }
+
+ @Override
+ public void send(String topic, EventMeshFrame frame, SendCallback callback) {
+ // no-op
+ }
+
+ @Override
+ public List<EventMeshFrame> poll(String topic, int partition, long startOffset, int maxEvents, long timeoutMs) {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void assignPartitions(String topic, List<Integer> partitions) {
+ // no-op
+ }
+
+ @Override
+ public void commitOffset(String topic, int partition, long offset) {
+ // no-op
+ }
+
+ @Override
+ public int partitionCount(String topic) {
+ return 0;
+ }
+
+ @Override
+ public boolean isStarted() {
+ return true;
+ }
+
+ @Override
+ public boolean isClosed() {
+ return false;
+ }
+
+ @Override
+ public void start() {
+ // no-op
+ }
+
+ @Override
+ public void shutdown() {
+ // no-op
+ }
+ }
+
+ private static UniRuntime runtime() {
+ return new UniRuntime(new StubStorage(), new InMemoryOffsetStore(),
+ 20L, 50L, 100, 50L, DeliveryTopology.LOCAL_STICKY_PULL, "test", "test:8080");
+ }
+
+ @Test
+ void enableClusterFlipsTopologyAndInjectsMeta() throws Exception {
+ // The core #5359 contract: after enableCluster, the runtime polls ONLY owned
+ // partitions (PARTITION_OWNED_PULL) through the injected shared MetaStore —
+ // not one ownership state machine in the app and a poll-all loop in the runtime.
+ UniRuntime runtime = runtime();
+ InMemoryMetaStore shared = new InMemoryMetaStore();
+ runtime.withClusterMeta(shared);
+ runtime.withTopology(DeliveryTopology.PARTITION_OWNED_PULL);
+ runtime.start();
+ try {
+ assertEquals(DeliveryTopology.PARTITION_OWNED_PULL, runtime.topology(),
+ "cluster mode must poll owned partitions only");
+ assertEquals(shared, runtime.clusterMeta(),
+ "the shared MetaStore must be wired into the runtime");
+ } finally {
+ runtime.shutdown();
+ }
+ }
+
+ @Test
+ void topologyCannotFlipAfterStart() throws Exception {
+ UniRuntime runtime = runtime();
+ runtime.start();
+ try {
+ IllegalStateException ex = assertThrows(IllegalStateException.class,
+ () -> runtime.withTopology(DeliveryTopology.PARTITION_OWNED_PULL));
+ assertTrue(ex.getMessage().contains("after start"),
+ "post-start topology change must fail fast, got: " + ex.getMessage());
+ } finally {
+ runtime.shutdown();
+ }
+ }
+
+ @Test
+ void clusterMetaCannotChangeAfterStart() throws Exception {
+ UniRuntime runtime = runtime();
+ runtime.start();
+ try {
+ IllegalStateException ex = assertThrows(IllegalStateException.class,
+ () -> runtime.withClusterMeta(new InMemoryMetaStore()));
+ assertTrue(ex.getMessage().contains("after start"),
+ "post-start meta change must fail fast, got: " + ex.getMessage());
+ } finally {
+ runtime.shutdown();
+ }
+ }
+
+ @Test
+ void nullTopologyRejected() {
+ UniRuntime runtime = runtime();
+ assertThrows(IllegalArgumentException.class,
+ () -> runtime.withTopology(null));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org