oscerd commented on code in PR #26733:
URL: https://github.com/apache/camel/pull/26733#discussion_r4080047067
##########
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:
Applied in ef9ab0c — you are right, and the comment was describing only one
of the three states.
Confirmed against the `AsyncEmbeddedEngine.close()` bytecode: it throws
`IllegalStateException` for `STARTING_TASKS` ("Cannot stop engine while tasks
are starting, this may lead to leaked resource."), `STOPPING` ("Engine is
already being shutting down.") and `STOPPED` ("Engine has been already shut
down."). Only the last is benign, so the level is now keyed on `engineStopped`
exactly as you suggested, and the comment says what the three states are.
_Claude Code on behalf of oscerd_
##########
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:
Agreed, and not adopting it for that reason.
`ScheduledPollConsumerHealthCheck` can start from `DOWN` only because
`consumer.isConsumerReady()` flips after the first successful poll; this
consumer has no equivalent signal, so `builder.state(initialState)` on the
`failure == null` path would leave the route `DOWN` forever wherever
`initialState=DOWN` is configured — a regression, not a consistency win.
`ConnectorCallback.taskStarted()` is indeed the right hook if startup
readiness is ever wanted. Noted for a follow-up rather than this PR.
_Claude Code on behalf of oscerd_
--
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]
