Andrea Cosentino created CAMEL-24889:
----------------------------------------
Summary: camel-debezium - the consumer is never told that the
embedded engine has failed
Key: CAMEL-24889
URL: https://issues.apache.org/jira/browse/CAMEL-24889
Project: Camel
Issue Type: Bug
Components: camel-debezium
Reporter: Andrea Cosentino
Assignee: Andrea Cosentino
h3. Summary
When the embedded Debezium engine dies, nothing in Camel is told. The route
stays {{Started}}, the
health check stays {{UP}}, and no change event is ever delivered again. The
consumer's own error
handling for this case cannot run.
h3. Details
{{DebeziumConsumer.doStart()}} submits the engine to a single-thread executor
and guards it:
{code:java}
executorService.submit(
() -> {
try {
dbzEngine.run();
} catch (Throwable e) {
LOG.error("Debezium engine has failed: {}", e.getMessage(), e);
}
});
{code}
That {{catch}} is unreachable. {{DebeziumEngine.create(Connect.class)}}
resolves through
{{META-INF/services/io.debezium.engine.DebeziumEngine$BuilderFactory}} to
{{ConvertingAsyncEngineBuilderFactory}}, i.e. an {{AsyncEmbeddedEngine}}.
Disassembling
{{AsyncEmbeddedEngine.run()}} from {{debezium-embedded}} shows the whole body
wrapped in its own
{{catch (Throwable)}} (exception table: {{from 2 to 196 target 204 Class
java/lang/Throwable}}),
which calls {{closeEngineWithException(t)}} - logging {{"Engine has failed
with"}} - and then
{{finishShutDown(t)}} -> {{callCompletionHandler(t)}}. {{run()}} never rethrows.
The only failure signal the engine offers is the {{CompletionCallback}} passed
to the builder, and
{{createDbzEngine()}} does not register one (nor a {{ConnectorCallback}}). So:
* the failure is invisible to Camel's error handling -
{{getExceptionHandler()}} is only ever called
from {{onEventListener}}, i.e. for a failure of the *route*, never for a
failure of the *engine*;
* it is invisible to health checks - {{ConsumerHealthCheck.doCallCheck}} only
consults a consumer
{{if (up && !external && consumer instanceof HealthCheckAware)}}, and
{{DebeziumConsumer}} does not
implement {{HealthCheckAware}}. The route reports UP while consuming nothing.
Any connector-level failure therefore produces a silently dead route: wrong
credentials, an
unreachable database, a missing table, a failed snapshot, or an offset store
that cannot be opened.
h3. Second symptom, same root cause
Once the engine has failed its state is {{STOPPED}}, and
{{AsyncEmbeddedEngine.close()}} throws on
that state ({{"Engine has been already shut down."}}).
{{DebeziumConsumer.doStop()}} calls it
unguarded:
{code:java}
protected void doStop() throws Exception {
if (dbzEngine != null) {
dbzEngine.close();
}
getEndpoint().getCamelContext().getExecutorServiceManager().shutdownGraceful(executorService);
super.doStop();
}
{code}
Stopping the route after the engine has died therefore throws, the
single-thread executor is never
shut down, and {{super.doStop()}} never runs.
h3. Proposed fix
* Register a {{DebeziumEngine.CompletionCallback}} and route a failure to
{{getExceptionHandler().handleException(...)}}, so the component honours the
normal Camel consumer
failure contract.
* Make {{DebeziumConsumer}} {{HealthCheckAware}} so that an engine that has
stopped with an error
takes the route's health check DOWN.
* Guard {{dbzEngine.close()}} so a stop after engine death does not skip the
rest of {{doStop()}}.
The existing {{FileConnectorEmbeddedDebeziumConfiguration}} test harness in
{{camel-debezium-common-component}} can drive a connector that fails to start,
so this is testable
without a database.
----
_Reported by Claude Code on behalf of oscerd (Andrea Cosentino)._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
