CAMEL-24919: report Debezium consumer startup readiness

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-db2-component.adoc

@@ -29,6 +29,10 @@ the application will resume from the last recorded offset,
 which may result in receiving duplicate events immediately after the restart.
 Therefore, your downstream routes should be tolerant enough of such a case and deduplicate events if needed.
 
+Before the engine starts polling, the consumer health check reports the configured initial state
+(`DOWN` by default). It reports `UP` once the connector tasks have started and the engine enters its polling phase,
+even if no change events have arrived. This does not indicate that an initial snapshot has completed.
+
 If the embedded engine stops with an error, for example because the connector cannot reach the database or
 the offset store cannot be read, it is not restarted: the route stays started but receives no further events.
 The failure is reported to the consumer exception handler and turns the route health check `DOWN`, so it can

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-mongodb-component.adoc

@@ -35,6 +35,10 @@ the application will resume from the last recorded offset,
 which may result in receiving duplicate events immediately after the restart.
 Therefore, your downstream routes should be tolerant enough of such a case and deduplicate events if needed.
 
+Before the engine starts polling, the consumer health check reports the configured initial state
+(`DOWN` by default). It reports `UP` once the connector tasks have started and the engine enters its polling phase,
+even if no change events have arrived. This does not indicate that an initial snapshot has completed.
+
 If the embedded engine stops with an error, for example because the connector cannot reach the database or
 the offset store cannot be read, it is not restarted: the route stays started but receives no further events.
 The failure is reported to the consumer exception handler and turns the route health check `DOWN`, so it can

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-mysql-component.adoc

@@ -28,6 +28,10 @@ the application will resume from the last recorded offset,
 which may result in receiving duplicate events immediately after the restart.
 Therefore, your downstream routes should be tolerant enough of such a case and deduplicate events if needed.
 
+Before the engine starts polling, the consumer health check reports the configured initial state
+(`DOWN` by default). It reports `UP` once the connector tasks have started and the engine enters its polling phase,
+even if no change events have arrived. This does not indicate that an initial snapshot has completed.
+
 If the embedded engine stops with an error, for example because the connector cannot reach the database or
 the offset store cannot be read, it is not restarted: the route stays started but receives no further events.
 The failure is reported to the consumer exception handler and turns the route health check `DOWN`, so it can

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-oracle-component.adoc

@@ -29,6 +29,10 @@ the application will resume from the last recorded offset,
 which may result in receiving duplicate events immediately after the restart.
 Therefore, your downstream routes should be tolerant enough of such a case and deduplicate events if needed.
 
+Before the engine starts polling, the consumer health check reports the configured initial state
+(`DOWN` by default). It reports `UP` once the connector tasks have started and the engine enters its polling phase,
+even if no change events have arrived. This does not indicate that an initial snapshot has completed.
+
 If the embedded engine stops with an error, for example because the connector cannot reach the database or
 the offset store cannot be read, it is not restarted: the route stays started but receives no further events.
 The failure is reported to the consumer exception handler and turns the route health check `DOWN`, so it can

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-postgres-component.adoc

@@ -29,6 +29,10 @@ the application will resume from the last recorded offset,
 which may result in receiving duplicate events immediately after the restart.
 Therefore, your downstream routes should be tolerant enough of such a case and deduplicate events if needed.
 
+Before the engine starts polling, the consumer health check reports the configured initial state
+(`DOWN` by default). It reports `UP` once the connector tasks have started and the engine enters its polling phase,
+even if no change events have arrived. This does not indicate that an initial snapshot has completed.
+
 If the embedded engine stops with an error, for example because the connector cannot reach the database or
 the offset store cannot be read, it is not restarted: the route stays started but receives no further events.
 The failure is reported to the consumer exception handler and turns the route health check `DOWN`, so it can

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-sqlserver-component.adoc

@@ -28,6 +28,10 @@ the application will resume from the last recorded offset,
 which may result in receiving duplicate events immediately after the restart.
 Therefore, your downstream routes should be tolerant enough of such a case and deduplicate events if needed.
 
+Before the engine starts polling, the consumer health check reports the configured initial state
+(`DOWN` by default). It reports `UP` once the connector tasks have started and the engine enters its polling phase,
+even if no change events have arrived. This does not indicate that an initial snapshot has completed.
+
 If the embedded engine stops with an error, for example because the connector cannot reach the database or
 the offset store cannot be read, it is not restarted: the route stays started but receives no further events.
 The failure is reported to the consumer exception handler and turns the route health check `DOWN`, so it can

components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumConsumer.java

@@ -42,6 +42,7 @@ public class DebeziumConsumer extends DefaultConsumer {
     private DebeziumEngine<ChangeEvent<SourceRecord, SourceRecord>> dbzEngine;
     private volatile Throwable engineFailure;
     private volatile boolean engineStopped;
+    private volatile boolean engineReady;
 
     public DebeziumConsumer(DebeziumEndpoint endpoint, Processor processor) {
         super(endpoint, processor);
@@ -59,11 +60,12 @@ protected void doBuild() throws Exception {
 
     @Override
     protected void doStart() throws Exception {
-        super.doStart();
-
+        engineReady = false;
         engineFailure = null;
         engineStopped = false;
 
+        super.doStart();
+
         // start a single threaded pool to monitor events
         executorService = endpoint.createExecutor(this);
 
@@ -115,10 +117,27 @@ Throwable getEngineFailure() {
         return engineFailure;
     }
 
+    boolean isEngineReady() {
+        return engineReady;
+    }
+
     private DebeziumEngine<ChangeEvent<SourceRecord, SourceRecord>> createDbzEngine() {
         return DebeziumEngine.create(Connect.class)
                 .using(configuration.createDebeziumConfiguration().asProperties())
                 .using(this::onEngineCompleted)
+                .using(new DebeziumEngine.ConnectorCallback() {
+                    @Override
+                    public void pollingStarted() {
+                        engineReady = true;
+                    }
+
+                    @Override
+                    public void pollingStopped() {
+                        // the engine is no longer polling, so it consumes no further change event, even when it
+                        // completed on its own without reporting a failure
+                        engineReady = false;
+                    }
+                })
                 .notifying(this::onEventListener)
                 .build();
     }

components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumConsumerHealthCheck.java

@@ -19,6 +19,7 @@
 import java.util.Map;
 
 import org.apache.camel.health.HealthCheck;
+import org.apache.camel.health.HealthCheckRegistry;
 import org.apache.camel.health.HealthCheckResultBuilder;
 import org.apache.camel.util.URISupport;
 
@@ -33,9 +34,12 @@ public class DebeziumConsumerHealthCheck implements HealthCheck {
     private final DebeziumConsumer consumer;
     private final String id;
     private final String sanitizedUri;
+    private final State initialState;
     private boolean enabled = true;
 
     public DebeziumConsumerHealthCheck(DebeziumConsumer consumer, String id) {
+        HealthCheckRegistry registry = HealthCheckRegistry.get(consumer.getEndpoint().getCamelContext());
+        this.initialState = registry != null ? registry.getInitialState() : State.DOWN;
         this.consumer = consumer;
         this.id = id;
         this.sanitizedUri = URISupport.sanitizeUri(consumer.getEndpoint().getEndpointUri());
@@ -76,8 +80,8 @@ public Result call(Map<String, Object> options) {
 
         final Throwable failure = consumer.getEngineFailure();
         if (failure == null) {
-            // the engine is either starting, running, or was stopped on request, none of which is a failure
-            return builder.up().build();
+            // Keep the configured initial state until the engine has started its polling tasks.
+            return builder.state(consumer.isEngineReady() ? State.UP : initialState).build();
         }
 
         builder.down();

components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumConsumerReadinessTest.java

@@ -0,0 +1,184 @@
+/*
+ * 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.debezium;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.debezium.configuration.FileConnectorEmbeddedDebeziumConfiguration;
+import org.apache.camel.health.HealthCheck;
+import org.apache.camel.health.HealthCheckRegistry;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.kafka.connect.connector.Task;
+import org.apache.kafka.connect.file.FileStreamSourceConnector;
+import org.apache.kafka.connect.file.FileStreamSourceTask;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+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.when;
+
+class DebeziumConsumerReadinessTest extends CamelTestSupport {
+    private static volatile CountDownLatch starting;
+    private static volatile CountDownLatch release;
+    private Path input;
+    private Path offsets;
+    private HealthCheckRegistry registry;
+
+    @ParameterizedTest
+    @EnumSource(HealthCheck.State.class)
+    void readinessWaitsForPollingAndResetsOnRestart(HealthCheck.State initialState) throws Exception {
+        when(registry.getInitialState()).thenReturn(initialState);
+        DebeziumConsumer consumer = (DebeziumConsumer) context.getRoute("readiness").getConsumer();
+        // Construct after setting the initial state, as with a consumer built during context startup.
+        consumer.setHealthCheck(new DebeziumConsumerHealthCheck(consumer, "consumer:readiness"));
+        HealthCheck check = consumer.getHealthCheck();
+        for (int i = 0; i < 2; i++) {
+            starting = new CountDownLatch(1);
+            release = new CountDownLatch(1);
+            consumer.start();
+            try {
+                assertTrue(starting.await(30, TimeUnit.SECONDS), "the source task should enter startup");
+                assertEquals(initialState, check.call().getState());
+                check.setEnabled(false);
+                assertEquals(HealthCheck.State.UNKNOWN, check.call().getState());
+                check.setEnabled(true);
+                release.countDown();
+                await().atMost(30, TimeUnit.SECONDS).until(consumer::isEngineReady);
+                await().atMost(30, TimeUnit.SECONDS)
+                        .untilAsserted(() -> assertEquals(HealthCheck.State.UP, check.call().getState()));
+                assertEquals(0, getMockEndpoint("mock:result").getReceivedCounter(),
+                        "readiness must not require a change event");
+            } finally {
+                release.countDown();
+                consumer.stop();
+            }
+        }
+    }
+
+    @Test
+    void theAutoCreatedHealthCheckReportsTheInitialStateUntilPolling() throws Exception {
+        DebeziumConsumer consumer = (DebeziumConsumer) context.getRoute("readiness").getConsumer();
+        starting = new CountDownLatch(1);
+        release = new CountDownLatch(1);
+        // no setHealthCheck() here: starting the consumer runs doBuild(), which creates the check itself
+        consumer.start();
+        try {
+            assertTrue(starting.await(30, TimeUnit.SECONDS), "the source task should enter startup");
+            HealthCheck check = consumer.getHealthCheck();
+            assertInstanceOf(DebeziumConsumerHealthCheck.class, check,
+                    "the consumer should create its own health check in doBuild()");
+            assertEquals("consumer:readiness", check.getId());
+            // DOWN is the registry initial state stubbed in createCamelContext, read when doBuild() ran
+            assertEquals(HealthCheck.State.DOWN, check.call().getState());
+            release.countDown();
+            await().atMost(30, TimeUnit.SECONDS)
+                    .untilAsserted(() -> assertEquals(HealthCheck.State.UP, check.call().getState()));
+        } finally {
+            release.countDown();
+            consumer.stop();
+        }
+        // the engine left its polling phase, so the consumer no longer reports itself as ready
+        assertFalse(consumer.isEngineReady(), "stopping the engine should clear readiness");
+    }
+
+    @AfterEach
+    void cleanup() throws Exception {
+        if (release != null) {
+            release.countDown();
+        }
+        context.stop();
+        Files.deleteIfExists(input);
+        Files.deleteIfExists(offsets);
+    }
+
+    @Override
+    protected CamelContext createCamelContext() throws Exception {
+        CamelContext context = super.createCamelContext();
+        registry = mock(HealthCheckRegistry.class);
+        when(registry.getInitialState()).thenReturn(HealthCheck.State.DOWN);
+        context.getCamelContextExtension().addContextPlugin(HealthCheckRegistry.class, registry);
+        input = Paths.get("target/data", "camel-debezium-readiness-input.txt").toAbsolutePath();
+        offsets = Paths.get("target/data", "camel-debezium-readiness-offset-store.dat").toAbsolutePath();
+        Files.createDirectories(input.getParent());
+        Files.deleteIfExists(input);
+        Files.deleteIfExists(offsets);
+        Files.createFile(input);
+        FileConnectorEmbeddedDebeziumConfiguration configuration = new FileConnectorEmbeddedDebeziumConfiguration() {
+            @Override
+            protected Class<?> configureConnectorClass() {
+                return DelayedConnector.class;
+            }
+        };
+        configuration.setName("readiness");
+        configuration.setTopicConfig("readiness");
+        configuration.setTestFilePath(input);
+        configuration.setOffsetStorageFileName(offsets.toString());
+        DebeziumTestComponent component = new DebeziumTestComponent(context);
+        component.setConfiguration(configuration);
+        context.addComponent("debezium", component);
+        context.disableJMX();
+        return context;
+    }
+
+    @Override
+    protected RoutesBuilder createRouteBuilder() {
+        return new RouteBuilder() {
+            @Override
+            public void configure() {
+                from("debezium:dummy").routeId("readiness").autoStartup(false).to("mock:result");
+            }
+        };
+    }
+
+    public static class DelayedConnector extends FileStreamSourceConnector {
+        @Override
+        public Class<? extends Task> taskClass() {
+            return DelayedTask.class;
+        }
+    }
+
+    public static class DelayedTask extends FileStreamSourceTask {
+        @Override
+        public void start(Map<String, String> props) {
+            starting.countDown();
+            try {
+                if (!release.await(30, TimeUnit.SECONDS)) {
+                    throw new IllegalStateException("Timed out waiting to release source task startup");
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new IllegalStateException(e);
+            }
+            super.start(props);
+        }
+    }
+}

components/camel-debezium/camel-debezium-db2/src/main/docs/debezium-db2-component.adoc

@@ -29,6 +29,10 @@ the application will resume from the last recorded offset,
 which may result in receiving duplicate events immediately after the restart.
 Therefore, your downstream routes should be tolerant enough of such a case and deduplicate events if needed.
 
+Before the engine starts polling, the consumer health check reports the configured initial state
+(`DOWN` by default). It reports `UP` once the connector tasks have started and the engine enters its polling phase,
+even if no change events have arrived. This does not indicate that an initial snapshot has completed.
+
 If the embedded engine stops with an error, for example because the connector cannot reach the database or
 the offset store cannot be read, it is not restarted: the route stays started but receives no further events.
 The failure is reported to the consumer exception handler and turns the route health check `DOWN`, so it can

components/camel-debezium/camel-debezium-mongodb/src/main/docs/debezium-mongodb-component.adoc

@@ -35,6 +35,10 @@ the application will resume from the last recorded offset,
 which may result in receiving duplicate events immediately after the restart.
 Therefore, your downstream routes should be tolerant enough of such a case and deduplicate events if needed.
 
+Before the engine starts polling, the consumer health check reports the configured initial state
+(`DOWN` by default). It reports `UP` once the connector tasks have started and the engine enters its polling phase,
+even if no change events have arrived. This does not indicate that an initial snapshot has completed.
+
 If the embedded engine stops with an error, for example because the connector cannot reach the database or
 the offset store cannot be read, it is not restarted: the route stays started but receives no further events.
 The failure is reported to the consumer exception handler and turns the route health check `DOWN`, so it can

components/camel-debezium/camel-debezium-mysql/src/main/docs/debezium-mysql-component.adoc

@@ -28,6 +28,10 @@ the application will resume from the last recorded offset,
 which may result in receiving duplicate events immediately after the restart.
 Therefore, your downstream routes should be tolerant enough of such a case and deduplicate events if needed.
 
+Before the engine starts polling, the consumer health check reports the configured initial state
+(`DOWN` by default). It reports `UP` once the connector tasks have started and the engine enters its polling phase,
+even if no change events have arrived. This does not indicate that an initial snapshot has completed.
+
 If the embedded engine stops with an error, for example because the connector cannot reach the database or
 the offset store cannot be read, it is not restarted: the route stays started but receives no further events.
 The failure is reported to the consumer exception handler and turns the route health check `DOWN`, so it can