davsclaus commented on code in PR #26733:
URL: https://github.com/apache/camel/pull/26733#discussion_r4070043068
##########
components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumConsumer.java:
##########
@@ -61,15 +76,23 @@ protected void doStart() throws Exception {
try {
dbzEngine.run();
} catch (Throwable e) {
- LOG.error("Debezium engine has failed: {}",
e.getMessage(), e);
+ // the engine reports its own failures through the
completion callback and is not
+ // expected to throw, so this is only a safety net
+ onEngineCompleted(false, e.getMessage(), e);
}
});
}
@Override
protected void doStop() throws Exception {
- if (dbzEngine != null) {
- dbzEngine.close();
+ if (dbzEngine != null && !engineStopped) {
+ try {
+ dbzEngine.close();
+ } catch (IllegalStateException e) {
+ // the engine refuses to be closed once it has stopped on its
own, which happens when it
+ // failed between the check above and this call, and then
there is nothing left to close
+ LOG.debug("Debezium engine was already stopped: {}",
e.getMessage());
+ }
Review Comment:
Question: this catch is broader than the comment describes. Debezium's
`close()` throws `IllegalStateException` for three states, not one: `STOPPED`
("Engine has been already shut down.", the benign case), but also "Cannot stop
engine while tasks are starting" and "Engine is already being shutting down".
In the tasks-starting case (route stopped during a slow snapshot start, or
context shutdown right after startup) the engine has **not** stopped, yet this
logs "already stopped" at DEBUG and `shutdownGraceful` then interrupts it.
Before this PR that case threw out of `doStop()`, so at least it was visible.
Suggest keying the log level on `engineStopped`, so the genuinely-benign
case stays quiet and the other two are visible:
```suggestion
} catch (IllegalStateException e) {
// the engine refuses to be closed once it has stopped on
its own, which happens when it
// failed between the check above and this call, and then
there is nothing left to close
if (engineStopped) {
LOG.debug("Debezium engine was already stopped: {}",
e.getMessage());
} else {
LOG.warn("Debezium engine could not be closed: {}",
e.getMessage());
}
}
```
##########
components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumConsumerHealthCheck.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * 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.util.Map;
+
+import org.apache.camel.health.HealthCheck;
+import org.apache.camel.health.HealthCheckResultBuilder;
+import org.apache.camel.util.URISupport;
+
+/**
+ * {@link HealthCheck} reporting the state of the embedded Debezium engine
that backs a {@link DebeziumConsumer}.
+ * <p>
+ * The engine runs on its own thread and does not restart itself, so once it
has stopped with an error the route no
+ * longer receives change events even though it is still started. This check
turns the route DOWN in that case.
+ */
+public class DebeziumConsumerHealthCheck implements HealthCheck {
+
+ private final DebeziumConsumer consumer;
+ private final String id;
+ private final String sanitizedUri;
+ private boolean enabled = true;
+
+ public DebeziumConsumerHealthCheck(DebeziumConsumer consumer, String id) {
+ this.consumer = consumer;
+ this.id = id;
+ this.sanitizedUri =
URISupport.sanitizeUri(consumer.getEndpoint().getEndpointUri());
+ }
+
+ @Override
+ public boolean isEnabled() {
+ return enabled;
+ }
+
+ @Override
+ public void setEnabled(boolean enabled) {
+ this.enabled = enabled;
+ }
+
+ @Override
+ public String getGroup() {
+ return "camel";
+ }
+
+ @Override
+ public String getId() {
+ return id;
+ }
+
+ @Override
+ public Result call(Map<String, Object> options) {
+ final HealthCheckResultBuilder builder =
HealthCheckResultBuilder.on(this);
+
+ // ensure to sanitize uri, so we do not show sensitive information
such as passwords
+ builder.detail(ENDPOINT_URI, sanitizedUri);
+
+ if (!isEnabled()) {
+ builder.message("Disabled");
+ builder.detail(CHECK_ENABLED, false);
+ return builder.unknown().build();
+ }
+
+ final Throwable failure = consumer.getEngineFailure();
Review Comment:
A caution on the `initialState` idea raised in the earlier review, in case
it gets picked up: `ScheduledPollConsumerHealthCheck` only gets away with
`initialState=DOWN` because `consumer.isConsumerReady()` flips after the first
successful poll. `DebeziumConsumer` has no equivalent "ready" signal, so
returning `builder.state(initialState)` here when `failure == null` would keep
the route `DOWN` forever under `initialState=DOWN`.
If startup readiness is ever wanted, Debezium's
`ConnectorCallback.taskStarted()` is the natural hook to flip a ready flag. Not
asking for it in this PR — the current `UP`-unless-failed semantics are
consistent with the stated scope.
##########
components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumConsumerEngineFailureTest.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import io.debezium.util.IoUtil;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+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.spi.ExceptionHandler;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The embedded engine runs on its own thread and reports a failure only
through its completion callback, so without
+ * that callback a connector that cannot start leaves the route started,
healthy and silent.
+ */
+public class DebeziumConsumerEngineFailureTest extends CamelTestSupport {
Review Comment:
CLAUDE.md test-visibility rule: new test classes must not be `public`.
```suggestion
class DebeziumConsumerEngineFailureTest extends CamelTestSupport {
```
##########
components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumConsumerEngineFailureTest.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import io.debezium.util.IoUtil;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+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.spi.ExceptionHandler;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The embedded engine runs on its own thread and reports a failure only
through its completion callback, so without
+ * that callback a connector that cannot start leaves the route started,
healthy and silent.
+ */
+public class DebeziumConsumerEngineFailureTest extends CamelTestSupport {
+
+ private static final String ROUTE_ID = "debezium-failing-engine";
+ private static final Path TEST_FILE_PATH
+ = Paths.get("target/data",
"camel-debezium-engine-failure-input.txt").toAbsolutePath();
+ private static final Path TEST_OFFSET_STORE_PATH
+ = Paths.get("target/data",
"camel-debezium-engine-failure-offset-store.txt").toAbsolutePath();
+
+ @BeforeEach
+ public void beforeEach() throws IOException {
Review Comment:
Same rule applies to `@BeforeEach`/`@AfterEach` methods.
```suggestion
void beforeEach() throws IOException {
```
##########
components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumConsumerEngineFailureTest.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import io.debezium.util.IoUtil;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+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.spi.ExceptionHandler;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The embedded engine runs on its own thread and reports a failure only
through its completion callback, so without
+ * that callback a connector that cannot start leaves the route started,
healthy and silent.
+ */
+public class DebeziumConsumerEngineFailureTest extends CamelTestSupport {
+
+ private static final String ROUTE_ID = "debezium-failing-engine";
+ private static final Path TEST_FILE_PATH
+ = Paths.get("target/data",
"camel-debezium-engine-failure-input.txt").toAbsolutePath();
+ private static final Path TEST_OFFSET_STORE_PATH
+ = Paths.get("target/data",
"camel-debezium-engine-failure-offset-store.txt").toAbsolutePath();
+
+ @BeforeEach
+ public void beforeEach() throws IOException {
+ IoUtil.createFile(TEST_FILE_PATH);
+ // an offset store the engine cannot read, which is what a corrupted
offset file looks like; the
+ // content must be long enough not to be mistaken for an empty store
+ Files.write(IoUtil.createFile(TEST_OFFSET_STORE_PATH).toPath(),
+ "this is not a serialized offset
store".getBytes(StandardCharsets.UTF_8));
+ }
+
+ @AfterEach
+ public void afterEach() throws IOException {
Review Comment:
```suggestion
void afterEach() throws IOException {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
