This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 8b9cbcd3748b camel-vertx: fail the start when the EventBus never
becomes ready (#29681)
8b9cbcd3748b is described below
commit 8b9cbcd3748b3acd5c36f014f4d6a8db03c1d12f
Author: dhairya05 <[email protected]>
AuthorDate: Fri Nov 13 12:53:42 2026 +0530
camel-vertx: fail the start when the EventBus never becomes ready (#29681)
Co-authored-by: Guillaume Nodet - AI Bot <[email protected]>
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/component/vertx/VertxComponent.java | 51 +++++-
.../camel/component/vertx/VertxStartupTest.java | 190 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 14 ++
3 files changed, 247 insertions(+), 8 deletions(-)
diff --git
a/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxComponent.java
b/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxComponent.java
index 9a9ab509514f..c2ab979c8b4d 100644
---
a/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxComponent.java
+++
b/components/camel-vertx/camel-vertx/src/main/java/org/apache/camel/component/vertx/VertxComponent.java
@@ -19,7 +19,9 @@ package org.apache.camel.component.vertx;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.VertxBuilder;
import io.vertx.core.VertxOptions;
@@ -162,34 +164,67 @@ public class VertxComponent extends DefaultComponent {
createdVertx = true;
final CountDownLatch latch = new CountDownLatch(1);
+ final AtomicReference<Throwable> failure = new AtomicReference<>();
+ // the callback writes here rather than to the field, so a build
that completes after the timeout
+ // cannot leave a half-published Vert.x behind
+ final AtomicReference<Vertx> created = new AtomicReference<>();
+ Future<Vertx> clusteredVertx = null;
// lets using a host / port if a host name is specified
if (clustered) {
+ if (timeout <= 0) {
+ throw new IllegalArgumentException(
+ "The timeout option must be a positive number of
seconds, was: " + timeout);
+ }
+
LOG.info("Creating Clustered Vertx {}:{}",
vertxOptions.getEventBusOptions().getHost(),
vertxOptions.getEventBusOptions().getPort());
// use the async api as we want to wait for the eventbus to be
ready before we are in started state
- vertxFactory.buildClustered().onComplete(event -> {
+ clusteredVertx = vertxFactory.buildClustered();
+ clusteredVertx.onComplete(event -> {
if (event.cause() != null) {
- LOG.warn("Error creating Clustered Vertx {}:{} due
{}", host, port,
- event.cause().getMessage(), event.cause());
+ failure.set(event.cause());
} else if (event.succeeded()) {
- vertx = event.result();
- LOG.info("EventBus is ready: {}", vertx);
+ created.set(event.result());
}
latch.countDown();
});
} else {
LOG.info("Creating Non-Clustered Vertx");
- vertx = vertxFactory.build();
- LOG.info("EventBus is ready: {}", vertx);
+ created.set(vertxFactory.build());
latch.countDown();
}
if (latch.getCount() > 0) {
LOG.info("Waiting for EventBus to be ready using {} sec as
timeout", timeout);
- latch.await(timeout, TimeUnit.SECONDS);
+ if (!latch.await(timeout, TimeUnit.SECONDS)) {
+ // the build is still running: a node that joins the
cluster after we gave up would keep its
+ // event-loop threads alive with nobody left to close it.
onSuccess also fires at once if the
+ // future completed between the await and this
registration, so no window is left open
+ clusteredVertx.onSuccess(v -> {
+ LOG.warn("Closing the Vert.x that became ready after
the timeout of {} seconds,"
+ + " as the component start has already
failed",
+ timeout);
+ v.close();
+ });
+ throw new IllegalStateException(
+ "The Vert.x EventBus was not ready within the
timeout of " + timeout + " seconds");
+ }
}
+
+ // the component must not reach the started state with a
half-built Vert.x: every endpoint created
+ // afterwards would dereference it
+ if (failure.get() != null) {
+ throw new IllegalStateException(
+ "Error creating Clustered Vertx " + host + ":" + port,
failure.get());
+ }
+ if (created.get() == null) {
+ throw new IllegalStateException("The Vert.x instance was not
created");
+ }
+
+ vertx = created.get();
+ LOG.info("EventBus is ready: {}", vertx);
} else {
LOG.debug("Using Vertx instance set on the component level.");
}
diff --git
a/components/camel-vertx/camel-vertx/src/test/java/org/apache/camel/component/vertx/VertxStartupTest.java
b/components/camel-vertx/camel-vertx/src/test/java/org/apache/camel/component/vertx/VertxStartupTest.java
new file mode 100644
index 000000000000..00b069447872
--- /dev/null
+++
b/components/camel-vertx/camel-vertx/src/test/java/org/apache/camel/component/vertx/VertxStartupTest.java
@@ -0,0 +1,190 @@
+/*
+ * 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.camel.component.vertx;
+
+import java.lang.reflect.Proxy;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.vertx.core.Future;
+import io.vertx.core.Promise;
+import io.vertx.core.Vertx;
+import io.vertx.core.VertxBuilder;
+import io.vertx.core.VertxOptions;
+import io.vertx.core.spi.VertxMetricsFactory;
+import io.vertx.core.spi.VertxTracerFactory;
+import io.vertx.core.spi.cluster.ClusterManager;
+import io.vertx.core.transport.Transport;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The component waits for the clustered EventBus before it reports itself
started, so a build that never finishes or
+ * that fails has to fail the start rather than leave the component without a
Vert.x instance.
+ */
+public class VertxStartupTest {
+
+ @Test
+ public void shouldFailWhenTheEventBusIsNotReadyWithinTheTimeout() throws
Exception {
+ try (DefaultCamelContext context = new DefaultCamelContext()) {
+ // a build that never completes, so the component waits out its
timeout
+ VertxComponent component = clusteredComponent(context,
Promise.<Vertx> promise().future());
+ component.setTimeout(1);
+
+ Exception e = assertThrows(Exception.class, component::start);
+
+ assertTrue(messages(e).contains("not ready within the timeout"),
messages(e));
+ assertNull(component.getVertx(), "the component must not be left
holding a half-built Vert.x");
+ }
+ }
+
+ @Test
+ public void shouldFailWhenBuildingTheClusteredVertxFails() throws
Exception {
+ try (DefaultCamelContext context = new DefaultCamelContext()) {
+ VertxComponent component = clusteredComponent(context,
+ Future.failedFuture(new IllegalStateException("no cluster
manager on the classpath")));
+
+ Exception e = assertThrows(Exception.class, component::start);
+
+ assertTrue(messages(e).contains("Error creating Clustered Vertx"),
messages(e));
+ assertTrue(messages(e).contains("no cluster manager on the
classpath"), messages(e));
+ assertNull(component.getVertx(), "the component must not be left
holding a half-built Vert.x");
+ }
+ }
+
+ @Test
+ public void shouldCloseAClusteredVertxThatArrivesAfterTheTimeout() throws
Exception {
+ final AtomicBoolean closed = new AtomicBoolean();
+ final Promise<Vertx> buildClustered = Promise.promise();
+
+ try (DefaultCamelContext context = new DefaultCamelContext()) {
+ VertxComponent component = clusteredComponent(context,
buildClustered.future());
+ component.setTimeout(1);
+
+ assertThrows(Exception.class, component::start);
+
+ // the cluster join finishes a moment after the component gave up
on it. A context-less promise runs its
+ // listeners on the completing thread, so close() has been called
by the time complete() returns
+ buildClustered.complete(recordingVertx(closed));
+
+ assertTrue(closed.get(), "a Vert.x that joins the cluster after
the timeout has to be closed again");
+ }
+ }
+
+ @Test
+ public void shouldRejectANonPositiveTimeout() throws Exception {
+ try (DefaultCamelContext context = new DefaultCamelContext()) {
+ VertxComponent component = clusteredComponent(context,
Promise.<Vertx> promise().future());
+ component.setTimeout(0);
+
+ Exception e = assertThrows(Exception.class, component::start);
+
+ assertTrue(messages(e).contains("must be a positive number of
seconds"), messages(e));
+ }
+ }
+
+ private static VertxComponent clusteredComponent(DefaultCamelContext
context, Future<Vertx> clustered) {
+ VertxComponent component = new VertxComponent();
+ component.setCamelContext(context);
+ // a host makes the component build a clustered Vert.x, which is the
asynchronous path
+ component.setHost("localhost");
+ component.setVertxFactory(new StubVertxBuilder(clustered));
+ return component;
+ }
+
+ /**
+ * A Vert.x that records whether it was closed. A dynamic proxy keeps this
to the one method the test cares about,
+ * rather than stubbing the whole interface.
+ */
+ private static Vertx recordingVertx(AtomicBoolean closed) {
+ return (Vertx) Proxy.newProxyInstance(Vertx.class.getClassLoader(),
new Class<?>[] { Vertx.class },
+ (proxy, method, args) -> {
+ switch (method.getName()) {
+ case "close":
+ closed.set(true);
+ return Future.succeededFuture();
+ case "toString":
+ return "recording Vertx";
+ case "hashCode":
+ return System.identityHashCode(proxy);
+ case "equals":
+ return proxy == args[0];
+ default:
+ return null;
+ }
+ });
+ }
+
+ private static String messages(Throwable t) {
+ StringBuilder sb = new StringBuilder();
+ for (Throwable current = t; current != null; current =
current.getCause()) {
+ sb.append(current).append(System.lineSeparator());
+ }
+ return sb.toString();
+ }
+
+ /**
+ * A builder whose clustered build is whatever the test hands it.
+ */
+ private static final class StubVertxBuilder implements VertxBuilder {
+
+ private final Future<Vertx> clustered;
+
+ private StubVertxBuilder(Future<Vertx> clustered) {
+ this.clustered = clustered;
+ }
+
+ @Override
+ public VertxBuilder with(VertxOptions options) {
+ return this;
+ }
+
+ @Override
+ public VertxBuilder withMetrics(VertxMetricsFactory factory) {
+ return this;
+ }
+
+ @Override
+ public VertxBuilder withTracer(VertxTracerFactory factory) {
+ return this;
+ }
+
+ @Override
+ public VertxBuilder withTransport(Transport transport) {
+ return this;
+ }
+
+ @Override
+ public VertxBuilder withClusterManager(ClusterManager clusterManager) {
+ return this;
+ }
+
+ @Override
+ public Vertx build() {
+ throw new UnsupportedOperationException("this test only builds a
clustered Vert.x");
+ }
+
+ @Override
+ public Future<Vertx> buildClustered() {
+ return clustered;
+ }
+ }
+
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 3c317079f677..8b6a06b99231 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -8489,3 +8489,17 @@ With `allowTemplateFromHeader=true`, a script from the
`CamelLanguageScript` hea
Before, with `cacheScript=true` it could replace the cached script of the
endpoint, so later messages without the header
ran it too. The header is now removed after use, so a later language endpoint
in the route runs its own script instead
of the header's. Set the header again before each language endpoint that
should use it.
+
+=== camel-vertx - the component fails to start when the clustered Vert.x is
not created
+
+When the component builds its own clustered Vert.x (the component `host` or
`port` option is set and no `vertxOptions` is
+given), it waits for the EventBus to be ready before it reports itself
started. That wait used to be ineffective: the
+result of the bounded wait was discarded and a failed build only logged a
warning, so the component started with no
+Vert.x instance and the first endpoint, producer or consumer to use it failed
with a `NullPointerException`.
+
+The component now fails to start, with the cause attached, when the clustered
Vert.x cannot be built or is not ready within
+the `timeout` option. An application whose cluster join is slower than
`timeout` (60 seconds by default) has to raise it.
+A Vert.x that completes after the timeout is closed, rather than being left
joined to the cluster with nobody to stop it.
+
+`timeout` must now be a positive number of seconds on that path: `timeout=0`
used to mean "do not wait", which left the
+component without a Vert.x, and is rejected with a clear message instead.
