[
https://issues.apache.org/jira/browse/CAMEL-24889?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18117944#comment-18117944
]
Andrea Cosentino commented on CAMEL-24889:
------------------------------------------
PR opened: https://github.com/apache/camel/pull/26733
The embedded engine never rethrows out of {{run()}} -
{{AsyncEmbeddedEngine.run()}} catches {{Throwable}} itself and reports through
the {{CompletionCallback}} - so the consumer now registers that callback, hands
the failure to the exception handler and reports {{DOWN}} from a new consumer
health check. Verified by reverting the fix: the new test then fails on the
exception-handler assertion.
----
_Claude Code on behalf of oscerd (Andrea Cosentino)._
> 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
> Priority: Major
>
> 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)
