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 e4925aa7d9..abfc27ddef 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 void enableCluster(org.apache.eventmesh.runtime.cluster.MetaStore metaSto // 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 445c34789e..490448a89f 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 0000000000..c699098e1b --- /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. + * + *

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).

+ */ +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 poll(String topic, int partition, long startOffset, int maxEvents, long timeoutMs) { + return Collections.emptyList(); + } + + @Override + public void assignPartitions(String topic, List 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)); + } +}