oscerd opened a new pull request, #26733:
URL: https://github.com/apache/camel/pull/26733
## What
`DebeziumEngine.create(Connect.class)` resolves through
`META-INF/services/io.debezium.engine.DebeziumEngine$BuilderFactory` to
`ConvertingAsyncEngineBuilderFactory`,
so the consumer runs an `AsyncEmbeddedEngine`. That engine's `run()` wraps
its whole body in its own
`catch (Throwable)`, logs the failure and reports it through the
`CompletionCallback` — it never rethrows.
`DebeziumConsumer` registered no such callback, so:
* the `catch (Throwable)` around `dbzEngine.run()` could never fire — it was
dead code;
* `getExceptionHandler()` was only ever called from `onEventListener`, i.e.
for a failure of the **route**,
never for a failure of the **engine**;
* `ConsumerHealthCheck` only consults a consumer that implements
`HealthCheckAware`, and this one did not.
A connector that cannot start — bad credentials, an unreachable database, a
missing table, a failed
snapshot, an unreadable offset store — therefore left the route **started
and healthy** while delivering
nothing, with no signal anywhere except a line in the Debezium logger.
The same root cause had a second symptom: once the engine has failed its
state is `STOPPED`, and
`close()` throws on that state (`"Engine has been already shut down."`).
`doStop()` called it unguarded,
so stopping such a route threw, the single-thread executor was never shut
down and `super.doStop()` never ran.
## Changes
* `DebeziumConsumer` registers a `CompletionCallback`. On failure it hands
the cause to
`getExceptionHandler()` and keeps it for the health check. A clean stop is
logged at DEBUG.
* New `DebeziumConsumerHealthCheck` (same shape as
`ScheduledPollConsumerHealthCheck`, `camel-api` types
only — no new dependency) reports `DOWN` with the engine error, `UP`
otherwise. Registered from
`doBuild()` as `consumer:<routeId>`, the same idiom
`ScheduledPollConsumer` uses.
* `doStop()` skips `close()` when the engine has already stopped, and
catches the `IllegalStateException`
for the race where it stops in between.
* The `catch (Throwable)` around `run()` is kept, but now routes into the
same failure path instead of
only logging — a safety net rather than dead code.
* The engine is still **not** restarted automatically; this change only
makes the failure observable.
## Tests
`DebeziumConsumerEngineFailureTest` points the existing file-connector
harness at an unreadable offset
store, which makes the engine fail with
`ConnectException: java.io.StreamCorruptedException: invalid stream header`,
and asserts the failure
reaches the consumer exception handler and the health check goes `DOWN`.
`DebeziumConsumerTest` additionally asserts a consumer with a live engine
stays `UP`.
Verified by reverting the fix and re-running: the new test then fails on
*"the engine failure should be handed to the consumer exception handler"*,
i.e. it genuinely catches the
reported bug rather than passing by construction. Full module suite green
(18 tests), full reactor
`mvn clean install -DskipTests -DskipITs` green.
## Docs
The "Note on handling failures" block in the six connector docs gains a
paragraph, and there is an
upgrade-guide entry for 4.23 because a deployment with a dead engine now
reports `DOWN` where it
previously reported `UP`.
_Claude Code on behalf of oscerd_
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
