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 41a9b22a6788 CAMEL-26419: camel-azure-cosmosdb - a readiness health
check for a consumer whose change feed processor has not started yet (#29045)
41a9b22a6788 is described below
commit 41a9b22a67888ebf77d9782c85e13ce86d6f7cad
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Nov 4 01:02:57 2026 +0530
CAMEL-26419: camel-azure-cosmosdb - a readiness health check for a consumer
whose change feed processor has not started yet (#29045)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../catalog/docs/azure-cosmosdb-component.adoc | 13 +-
.../camel-azure/camel-azure-cosmosdb/pom.xml | 5 +
.../src/main/docs/azure-cosmosdb-component.adoc | 13 +-
.../component/azure/cosmosdb/CosmosDbConsumer.java | 62 ++++++-
.../cosmosdb/CosmosDbConsumerHealthCheck.java | 87 +++++++++
.../CosmosDbConsumerHealthCheckRegistryTest.java | 194 ++++++++++++++++++++
.../cosmosdb/CosmosDbConsumerHealthCheckTest.java | 203 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 5 +
8 files changed, 571 insertions(+), 11 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-cosmosdb-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-cosmosdb-component.adoc
index dbfffa5fbdee..9234de22d05e 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-cosmosdb-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-cosmosdb-component.adoc
@@ -74,6 +74,18 @@ This component implements the async Consumer and producer.
This allows camel route to consume and produce events asynchronously without
blocking any threads.
+=== Consumer health check
+
+When Cosmos DB cannot be reached yet, the lease container does not exist yet
or a role assignment has not propagated
+yet, the change feed processor of a consumer cannot start. The consumer then
tries to start it again, with a backoff of
+1 to 30 seconds, until it starts or the consumer stops, and meanwhile the
route and the consumer are started. With the
+consumer health checks of Camel enabled, the readiness check of the consumer
of the route (`consumer:<routeId>`) is
+the initial state of the health checks (`camel.health.initial-state`, DOWN by
default) while the first attempt runs,
+DOWN with the failure of the last attempt and the number of failed attempts
(`failed.starts`) while it cannot start,
+and UP once it has started. It does not follow what the change feed processor
does after it started. As for any
+consumer check, camel-health asks it only for a route that starts with the
CamelContext, and
+`camel.health.exclude-pattern=consumer:<routeId>` leaves it out.
+
=== Message headers evaluated by the component producer
[width="100%",cols="10%,10%,10%,70%",options="header",]
|=======================================================================
@@ -893,7 +905,6 @@ YAML::
----
====
-
== Important Development Notes
When developing on this component, you will need to obtain your Azure
accessKey in order to run the integration tests. In addition to the mocked unit
tests,
diff --git a/components/camel-azure/camel-azure-cosmosdb/pom.xml
b/components/camel-azure/camel-azure-cosmosdb/pom.xml
index 38c4c04e71b4..d733962c4ba9 100644
--- a/components/camel-azure/camel-azure-cosmosdb/pom.xml
+++ b/components/camel-azure/camel-azure-cosmosdb/pom.xml
@@ -68,6 +68,11 @@
<artifactId>camel-test-junit6</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-health</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
diff --git
a/components/camel-azure/camel-azure-cosmosdb/src/main/docs/azure-cosmosdb-component.adoc
b/components/camel-azure/camel-azure-cosmosdb/src/main/docs/azure-cosmosdb-component.adoc
index dbfffa5fbdee..9234de22d05e 100644
---
a/components/camel-azure/camel-azure-cosmosdb/src/main/docs/azure-cosmosdb-component.adoc
+++
b/components/camel-azure/camel-azure-cosmosdb/src/main/docs/azure-cosmosdb-component.adoc
@@ -74,6 +74,18 @@ This component implements the async Consumer and producer.
This allows camel route to consume and produce events asynchronously without
blocking any threads.
+=== Consumer health check
+
+When Cosmos DB cannot be reached yet, the lease container does not exist yet
or a role assignment has not propagated
+yet, the change feed processor of a consumer cannot start. The consumer then
tries to start it again, with a backoff of
+1 to 30 seconds, until it starts or the consumer stops, and meanwhile the
route and the consumer are started. With the
+consumer health checks of Camel enabled, the readiness check of the consumer
of the route (`consumer:<routeId>`) is
+the initial state of the health checks (`camel.health.initial-state`, DOWN by
default) while the first attempt runs,
+DOWN with the failure of the last attempt and the number of failed attempts
(`failed.starts`) while it cannot start,
+and UP once it has started. It does not follow what the change feed processor
does after it started. As for any
+consumer check, camel-health asks it only for a route that starts with the
CamelContext, and
+`camel.health.exclude-pattern=consumer:<routeId>` leaves it out.
+
=== Message headers evaluated by the component producer
[width="100%",cols="10%,10%,10%,70%",options="header",]
|=======================================================================
@@ -893,7 +905,6 @@ YAML::
----
====
-
== Important Development Notes
When developing on this component, you will need to obtain your Azure
accessKey in order to run the integration tests. In addition to the mocked unit
tests,
diff --git
a/components/camel-azure/camel-azure-cosmosdb/src/main/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConsumer.java
b/components/camel-azure/camel-azure-cosmosdb/src/main/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConsumer.java
index 8e306e010af1..cf58b43cbb0f 100644
---
a/components/camel-azure/camel-azure-cosmosdb/src/main/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConsumer.java
+++
b/components/camel-azure/camel-azure-cosmosdb/src/main/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConsumer.java
@@ -61,8 +61,14 @@ public class CosmosDbConsumer extends DefaultConsumer {
// the start() of changeFeedProcessor has not completed yet
private boolean starting;
private Disposable startSubscription;
- // the number of attempts to start the change feed processor that failed
(logged when it starts)
+ // the number of attempts to start the change feed processor that failed
(logged when it starts, and reported by the
+ // health check)
private volatile long failedStarts;
+ // the failure of the last attempt to start the change feed processor of
this start of the consumer, for the health
+ // check
+ private volatile Throwable lastStartFailure;
+ // the change feed processor of this start of the consumer has started,
for the health check
+ private volatile boolean changeFeedProcessorStarted;
public CosmosDbConsumer(final CosmosDbEndpoint endpoint, final Processor
processor) {
super(endpoint, processor);
@@ -89,7 +95,15 @@ public class CosmosDbConsumer extends DefaultConsumer {
@Override
protected void doStart() throws Exception {
super.doStart();
+ // cleared in the reverse order of onStartAttemptFailed, as in doStop
+ lastStartFailure = null;
failedStarts = 0;
+ changeFeedProcessorStarted = false;
+ if (getHealthCheck() == null) {
+ // readiness: not UP until the change feed processor has started
(used when the consumer health checks of
+ // Camel are enabled)
+ setHealthCheck(new CosmosDbConsumerHealthCheck(this,
getRouteId()));
+ }
final long startGeneration;
synchronized (lock) {
startGeneration = ++generation;
@@ -111,10 +125,8 @@ public class CosmosDbConsumer extends DefaultConsumer {
// before the attempts were retried (so
bridgeErrorHandler routes it once, not once per
// attempt); then one line per failed attempt, as it
is tried again for as long as it fails:
// the stack trace at DEBUG
- .doBeforeRetry(signal -> {
- failedStarts = signal.totalRetries() + 1;
- onStartAttemptFailed(startGeneration,
signal.totalRetries() + 1, signal.failure());
- }))
+ .doBeforeRetry(signal ->
onStartAttemptFailed(startGeneration, signal.totalRetries() + 1,
+ signal.failure())))
.doOnSuccess(aVoid -> {
if (failedStarts > 0) {
LOG.info("Started the change feed processor of {}
after {} failed attempt(s)",
@@ -132,10 +144,17 @@ public class CosmosDbConsumer extends DefaultConsumer {
private void onStartAttemptFailed(final long startGeneration, final long
attempt, final Throwable failure) {
String uri = URISupport.sanitizeUri(getEndpoint().getEndpointUri());
- if (isStale(startGeneration)) {
- // the consumer stopped after this attempt failed: not a failure
to report (a next attempt does nothing)
- LOG.debug("The change feed processor of {} could not start before
the consumer stopped", uri, failure);
- return;
+ synchronized (lock) {
+ if (startGeneration != generation) {
+ // the consumer stopped after this attempt failed: not a
failure to report (a next attempt does nothing)
+ LOG.debug("The change feed processor of {} could not start
before the consumer stopped", uri, failure);
+ return;
+ }
+ // under the lock, as doStop clears them. The count before the
failure: the health check reads the failure
+ // first and the count second, so a failure it sees always comes
with its count (it can see a count without
+ // its failure yet, which only reports the start as running)
+ failedStarts = attempt;
+ lastStartFailure = failure;
}
if (attempt == 1) {
try {
@@ -210,6 +229,7 @@ public class CosmosDbConsumer extends DefaultConsumer {
synchronized (lock) {
if (startGeneration == generation) {
starting = false;
+ changeFeedProcessorStarted = true;
sink.success();
return;
}
@@ -268,6 +288,9 @@ public class CosmosDbConsumer extends DefaultConsumer {
// any attempt to start that is still running is of an older
generation now: it does not start a processor
// (or stops it when its start ends, if this comes just as it
calls start())
generation++;
+ changeFeedProcessorStarted = false;
+ lastStartFailure = null;
+ failedStarts = 0;
subscription = startSubscription;
startSubscription = null;
processor = changeFeedProcessor;
@@ -297,6 +320,27 @@ public class CosmosDbConsumer extends DefaultConsumer {
super.doStop();
}
+ /**
+ * Whether the change feed processor of this start of the consumer has
started.
+ */
+ boolean isChangeFeedProcessorStarted() {
+ return changeFeedProcessorStarted;
+ }
+
+ /**
+ * The number of attempts to start the change feed processor that failed
since the consumer started.
+ */
+ long getFailedStarts() {
+ return failedStarts;
+ }
+
+ /**
+ * The failure of the last attempt to start the change feed processor
since the consumer started, or null.
+ */
+ Throwable getLastStartFailure() {
+ return lastStartFailure;
+ }
+
public CosmosDbConfiguration getConfiguration() {
return getEndpoint().getConfiguration();
}
diff --git
a/components/camel-azure/camel-azure-cosmosdb/src/main/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConsumerHealthCheck.java
b/components/camel-azure/camel-azure-cosmosdb/src/main/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConsumerHealthCheck.java
new file mode 100644
index 000000000000..b8f6150701e0
--- /dev/null
+++
b/components/camel-azure/camel-azure-cosmosdb/src/main/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConsumerHealthCheck.java
@@ -0,0 +1,87 @@
+/*
+ * 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.azure.cosmosdb;
+
+import java.util.Map;
+
+import org.apache.camel.health.HealthCheck;
+import org.apache.camel.health.HealthCheckRegistry;
+import org.apache.camel.health.HealthCheckResultBuilder;
+
+/**
+ * Readiness health check of the Azure Cosmos DB consumer: the initial state
of the health checks (DOWN by default)
+ * while the first attempt to start the change feed processor runs, DOWN with
the last start failure and the number of
+ * failed attempts while it cannot start (the consumer tries again with a
backoff until it starts), and UP once it has
+ * started.
+ */
+public class CosmosDbConsumerHealthCheck implements HealthCheck {
+
+ private final CosmosDbConsumer consumer;
+ private final String routeId;
+ private final State initialState;
+ private volatile boolean enabled = true;
+
+ public CosmosDbConsumerHealthCheck(CosmosDbConsumer consumer, String
routeId) {
+ this.consumer = consumer;
+ this.routeId = routeId;
+ // camel.health.initial-state, as the Debezium consumer check does
+ HealthCheckRegistry registry =
HealthCheckRegistry.get(consumer.getEndpoint().getCamelContext());
+ this.initialState = registry != null ? registry.getInitialState() :
State.DOWN;
+ }
+
+ @Override
+ public String getGroup() {
+ return "camel";
+ }
+
+ @Override
+ public String getId() {
+ return routeId != null ? "consumer:azure-cosmosdb-" + routeId :
"consumer:azure-cosmosdb";
+ }
+
+ @Override
+ public boolean isEnabled() {
+ return enabled;
+ }
+
+ @Override
+ public void setEnabled(boolean enabled) {
+ this.enabled = enabled;
+ }
+
+ @Override
+ public Result call(Map<String, Object> options) {
+ HealthCheckResultBuilder builder = HealthCheckResultBuilder.on(this);
+ if (!enabled) {
+ // as AbstractHealthCheck of camel-health does for a disabled check
+ return builder.message("Disabled").unknown().build();
+ }
+ if (consumer.isChangeFeedProcessorStarted()) {
+ return builder.up().build();
+ }
+ Throwable failure = consumer.getLastStartFailure();
+ long failedStarts = consumer.getFailedStarts();
+ builder.detail("failed.starts", failedStarts);
+ if (failure == null) {
+ // the first attempt to start runs
+ return builder.state(initialState).message("The change feed
processor is starting").build();
+ }
+ return builder.down().error(failure)
+ .message("The change feed processor has not started yet (" +
failedStarts + " failed attempt(s))")
+ .build();
+ }
+}
diff --git
a/components/camel-azure/camel-azure-cosmosdb/src/test/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConsumerHealthCheckRegistryTest.java
b/components/camel-azure/camel-azure-cosmosdb/src/test/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConsumerHealthCheckRegistryTest.java
new file mode 100644
index 000000000000..795931fb8f63
--- /dev/null
+++
b/components/camel-azure/camel-azure-cosmosdb/src/test/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConsumerHealthCheckRegistryTest.java
@@ -0,0 +1,194 @@
+/*
+ * 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.azure.cosmosdb;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.azure.cosmos.ChangeFeedProcessor;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.health.HealthCheck;
+import org.apache.camel.health.HealthCheckHelper;
+import org.apache.camel.health.HealthCheckRegistry;
+import org.apache.camel.impl.health.ConsumersHealthCheckRepository;
+import org.apache.camel.impl.health.DefaultHealthCheckRegistry;
+import org.apache.camel.support.DefaultComponent;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import reactor.core.publisher.Mono;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * The health check of the consumer as camel-health reports it: the readiness
check of the consumer of the route
+ * ({@code consumer:<routeId>}) is DOWN, with the start failure, until the
change feed processor has started, and is not
+ * a liveness check.
+ */
+class CosmosDbConsumerHealthCheckRegistryTest extends CamelTestSupport {
+
+ private final AtomicBoolean reachable = new AtomicBoolean();
+ // completes the first attempt to start (with a failure) when the test
lets it
+ private final CompletableFuture<Void> firstAttempt = new
CompletableFuture<>();
+ private final AtomicInteger attempts = new AtomicInteger();
+
+ private ChangeFeedProcessor processor() {
+ ChangeFeedProcessor processor = mock(ChangeFeedProcessor.class);
+ AtomicBoolean started = new AtomicBoolean();
+ when(processor.isStarted()).thenAnswer(invocation -> started.get());
+ when(processor.stop()).thenAnswer(invocation -> {
+ if (!started.compareAndSet(true, false)) {
+ throw new IllegalStateException("The ChangeFeedProcessor
instance has not fully started");
+ }
+ return Mono.empty();
+ });
+ when(processor.start()).thenAnswer(invocation -> {
+ if (attempts.incrementAndGet() == 1) {
+ return Mono.fromFuture(firstAttempt);
+ }
+ if (!reachable.get()) {
+ return Mono.error(new IllegalStateException("Resource Not
Found: lease container (test)"));
+ }
+ started.set(true);
+ return Mono.empty();
+ });
+ return processor;
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ DefaultHealthCheckRegistry registry = new
DefaultHealthCheckRegistry(context);
+
context.getCamelContextExtension().addContextPlugin(HealthCheckRegistry.class,
registry);
+ registry.register(new ConsumersHealthCheckRepository());
+ // an azure-cosmosdb endpoint whose consumer uses the change feed
processor of the test
+ context.addComponent("testcosmos", new DefaultComponent() {
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) throws Exception {
+ parameters.clear();
+ CosmosDbEndpoint endpoint = (CosmosDbEndpoint)
getCamelContext().getComponent("azure-cosmosdb")
+ .createEndpoint(
+ "azure-cosmosdb:" + remaining +
"?databaseEndpoint=https://test.com:443&accountKey=myKey";);
+ return new CosmosDbEndpoint(uri, endpoint.getComponent(),
endpoint.getConfiguration()) {
+ @Override
+ public void doStart() {
+ // no Cosmos DB client: the consumer uses the change
feed processor of the test
+ }
+
+ @Override
+ public void doStop() {
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws
Exception {
+ CosmosDbConsumer consumer = new CosmosDbConsumer(this,
processor) {
+ @Override
+ ChangeFeedProcessor createChangeFeedProcessor() {
+ return processor();
+ }
+ };
+ configureConsumer(consumer);
+ return consumer;
+ }
+ };
+ }
+ });
+ return context;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("testcosmos:mydb/myContainer").routeId("cosmos").to("mock:result");
+ }
+ };
+ }
+
+ private static Optional<HealthCheck.Result>
consumerCheck(Iterable<HealthCheck.Result> results) {
+ for (HealthCheck.Result result : results) {
+ if ("consumer:cosmos".equals(result.getCheck().getId())) {
+ return Optional.of(result);
+ }
+ }
+ return Optional.empty();
+ }
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ private CosmosDbConsumer startRoute(HealthCheck.State initialState) throws
Exception {
+
context.getCamelContextExtension().getContextPlugin(HealthCheckRegistry.class).setInitialState(
+ initialState != null ? initialState : HealthCheck.State.DOWN);
+ context.addRoutes(createRouteBuilder());
+ context.start();
+ return (CosmosDbConsumer) context.getRoute("cosmos").getConsumer();
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = HealthCheck.State.class, names = { "DOWN", "UP" })
+ void readinessIsTheInitialStateWhileTheFirstAttemptRuns(HealthCheck.State
initialState) throws Exception {
+ CosmosDbConsumer consumer = startRoute(initialState);
+ await().atMost(20, TimeUnit.SECONDS).until(() -> attempts.get() == 1);
+
+ // camel.health.initial-state, as for the Debezium consumer
+ assertEquals(initialState,
consumerCheck(HealthCheckHelper.invokeReadiness(context)).orElseThrow().getState());
+
+ // the first attempt fails: DOWN
+ firstAttempt.completeExceptionally(new IllegalStateException("Resource
Not Found: lease container (test)"));
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
consumer.getLastStartFailure() != null);
+ assertEquals(HealthCheck.State.DOWN,
+
consumerCheck(HealthCheckHelper.invokeReadiness(context)).orElseThrow().getState());
+ }
+
+ @Test
+ void readinessIsDownUntilTheChangeFeedProcessorStarted() throws Exception {
+ CosmosDbConsumer consumer = startRoute(null);
+ firstAttempt.completeExceptionally(new IllegalStateException("Resource
Not Found: lease container (test)"));
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
consumer.getLastStartFailure() != null);
+
+ HealthCheck.Result down =
consumerCheck(HealthCheckHelper.invokeReadiness(context)).orElseThrow();
+ assertEquals(HealthCheck.State.DOWN, down.getState());
+ assertEquals("Resource Not Found: lease container (test)",
down.getError().orElseThrow().getMessage());
+ assertTrue(((Number)
down.getDetails().get("failed.starts")).longValue() >= 1,
down.getDetails().toString());
+ // not a liveness check (camel-health only asks the consumer check for
readiness)
+ assertFalse(consumerCheck(HealthCheckHelper.invokeLiveness(context))
+ .filter(result -> result.getState() ==
HealthCheck.State.DOWN).isPresent());
+
+ reachable.set(true);
+ await().atMost(20, TimeUnit.SECONDS).until(
+ () ->
consumerCheck(HealthCheckHelper.invokeReadiness(context)).orElseThrow().getState()
+ == HealthCheck.State.UP);
+ }
+}
diff --git
a/components/camel-azure/camel-azure-cosmosdb/src/test/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConsumerHealthCheckTest.java
b/components/camel-azure/camel-azure-cosmosdb/src/test/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConsumerHealthCheckTest.java
new file mode 100644
index 000000000000..3019813c2b5f
--- /dev/null
+++
b/components/camel-azure/camel-azure-cosmosdb/src/test/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConsumerHealthCheckTest.java
@@ -0,0 +1,203 @@
+/*
+ * 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.azure.cosmosdb;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+
+import com.azure.cosmos.ChangeFeedProcessor;
+import org.apache.camel.health.HealthCheck;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Mono;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * The readiness health check of the consumer: DOWN while the change feed
processor cannot start (with the last failure
+ * and the number of failed attempts), UP once it has started, and back to its
initial state after a stop.
+ */
+class CosmosDbConsumerHealthCheckTest extends CamelTestSupport {
+
+ @Test
+ void downWhileTheChangeFeedProcessorCannotStartThenUp() throws Exception {
+ AtomicBoolean reachable = new AtomicBoolean();
+ AtomicInteger starts = new AtomicInteger();
+ ChangeFeedProcessor processor = processor(reachable, starts);
+ CosmosDbConsumer consumer = consumer(processor);
+
+ consumer.start();
+ try {
+ HealthCheck check = consumer.getHealthCheck();
+ assertInstanceOf(CosmosDbConsumerHealthCheck.class, check);
+
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
consumer.getLastStartFailure() != null);
+ HealthCheck.Result down = check.call();
+ assertEquals(HealthCheck.State.DOWN, down.getState());
+ assertTrue(down.getError().isPresent());
+ assertEquals("Resource Not Found: lease container (test)",
down.getError().get().getMessage());
+ assertTrue(((Number)
down.getDetails().get("failed.starts")).longValue() >= 1,
down.getDetails().toString());
+ assertTrue(down.getMessage().orElse("").startsWith("The change
feed processor has not started yet"),
+ down.getMessage().toString());
+
+ reachable.set(true);
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
check.call().getState() == HealthCheck.State.UP);
+ } finally {
+ consumer.stop();
+ }
+
+ // after a stop: no failure of the stopped start, and not started
+ HealthCheck.Result stopped = consumer.getHealthCheck().call();
+ assertEquals(HealthCheck.State.DOWN, stopped.getState());
+ assertFalse(stopped.getError().isPresent());
+ assertEquals(0L, ((Number)
stopped.getDetails().get("failed.starts")).longValue());
+ assertFalse(consumer.isChangeFeedProcessorStarted());
+ }
+
+ @Test
+ void aStartThatEndsAfterARestartDoesNotMakeItUp() throws Exception {
+ // the change feed processor of the first start is still starting when
the consumer is stopped and started
+ // again; the one of the second start cannot start. When the first
one's start ends, the check stays DOWN
+ CompletableFuture<Void> firstStart = new CompletableFuture<>();
+ ChangeFeedProcessor first = mock(ChangeFeedProcessor.class);
+ when(first.start()).thenReturn(Mono.fromFuture(firstStart));
+ when(first.isStarted()).thenReturn(false);
+ AtomicInteger starts = new AtomicInteger();
+ ChangeFeedProcessor second = processor(new AtomicBoolean(), starts);
+ Deque<ChangeFeedProcessor> processors = new
ArrayDeque<>(List.of(first, second));
+ CosmosDbConsumer consumer = consumer(() -> processors.size() > 1 ?
processors.poll() : processors.peek());
+
+ consumer.start();
+ try {
+ verify(first, timeout(20000)).start();
+ consumer.stop();
+ consumer.start();
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
consumer.getLastStartFailure() != null);
+
+ firstStart.complete(null);
+ // the attempt of the first start sees that the consumer stopped
meanwhile, and stops that processor
+ verify(first, timeout(20000).atLeast(2)).isStarted();
+ HealthCheck.Result result = consumer.getHealthCheck().call();
+ assertEquals(HealthCheck.State.DOWN, result.getState());
+ assertFalse(consumer.isChangeFeedProcessorStarted());
+ } finally {
+ consumer.stop();
+ }
+ }
+
+ @Test
+ void aDisabledCheckIsUnknown() throws Exception {
+ CosmosDbConsumer consumer = consumer(processor(new AtomicBoolean(),
new AtomicInteger()));
+ consumer.start();
+ try {
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
consumer.getLastStartFailure() != null);
+ consumer.getHealthCheck().setEnabled(false);
+ HealthCheck.Result result = consumer.getHealthCheck().call();
+ assertEquals(HealthCheck.State.UNKNOWN, result.getState());
+ assertEquals("Disabled", result.getMessage().orElse(null));
+ } finally {
+ consumer.stop();
+ }
+ }
+
+ @Test
+ void upOnceTheFirstStartSucceeds() throws Exception {
+ AtomicInteger starts = new AtomicInteger();
+ ChangeFeedProcessor processor = processor(new AtomicBoolean(true),
starts);
+ CosmosDbConsumer consumer = consumer(processor);
+
+ consumer.start();
+ try {
+ await().atMost(20, TimeUnit.SECONDS)
+ .until(() -> consumer.getHealthCheck().call().getState()
== HealthCheck.State.UP);
+ assertEquals(0, consumer.getFailedStarts());
+ assertEquals(1, starts.get());
+ } finally {
+ consumer.stop();
+ }
+
+ // started again: UP again once its change feed processor started
+ consumer.start();
+ try {
+ await().atMost(20, TimeUnit.SECONDS)
+ .until(() -> consumer.getHealthCheck().call().getState()
== HealthCheck.State.UP);
+ assertEquals(2, starts.get());
+ } finally {
+ consumer.stop();
+ }
+ }
+
+ /**
+ * A change feed processor that cannot start while reachable is false (as
when the lease container does not exist
+ * yet), and starts once it is true.
+ */
+ private static ChangeFeedProcessor processor(AtomicBoolean reachable,
AtomicInteger starts) {
+ ChangeFeedProcessor processor = mock(ChangeFeedProcessor.class);
+ AtomicBoolean started = new AtomicBoolean();
+ when(processor.isStarted()).thenAnswer(invocation -> started.get());
+ when(processor.stop()).thenAnswer(invocation -> {
+ if (!started.compareAndSet(true, false)) {
+ throw new IllegalStateException("The ChangeFeedProcessor
instance has not fully started");
+ }
+ return Mono.empty();
+ });
+ when(processor.start()).thenAnswer(invocation -> {
+ starts.incrementAndGet();
+ if (!reachable.get()) {
+ return Mono.error(new IllegalStateException("Resource Not
Found: lease container (test)"));
+ }
+ started.set(true);
+ return Mono.empty();
+ });
+ return processor;
+ }
+
+ private CosmosDbConsumer consumer(ChangeFeedProcessor processor) throws
Exception {
+ return consumer(() -> processor);
+ }
+
+ private CosmosDbConsumer consumer(Supplier<ChangeFeedProcessor>
processors) throws Exception {
+ Map<String, Object> params = new HashMap<>();
+ params.put("databaseEndpoint", "https://test.com:443";);
+ params.put("accountKey", "myKey");
+ CosmosDbEndpoint endpoint = (CosmosDbEndpoint)
context.getComponent("azure-cosmosdb", CosmosDbComponent.class)
+ .createEndpoint("azure-cosmosdb://mydb/myContainer",
"mydb/myContainer", params);
+ return new CosmosDbConsumer(endpoint, exchange -> {
+ }) {
+ @Override
+ ChangeFeedProcessor createChangeFeedProcessor() {
+ return processors.get();
+ }
+ };
+ }
+}
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 84ba80b8ff78..4520961b6bcf 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
@@ -1119,6 +1119,11 @@ never received a change. The first failure of each start
still goes to the `Exce
`bridgeErrorHandler=true` it is routed to the error handler once), and each
further failed attempt is logged as one
WARN line.
+Its readiness health check (`consumer:<routeId>`, with the consumer health
checks of Camel enabled, the default) is now
+the initial state (`camel.health.initial-state`, DOWN by default) while the
first attempt to start runs, and DOWN while
+the change feed processor cannot start, with the failure of the last attempt;
before, it was UP. To leave
+the check out, as before, set
`camel.health.exclude-pattern=consumer:<routeId>`.
+
=== camel-vertx-websocket - consumers on the same host and port must use the
same SSLContextParameters
All `vertx-websocket` consumers on the same host and port are served by one
HTTP server, which is created from the
