debezium/dbz#1662 Start & Stop API for the Quarkus Runtime Extension

[yuanglili]

Summary

Adds start(EngineManifest) and stop(EngineManifest) methods to DebeziumConnectorRegistry, enabling users to manually control engine lifecycle.

Changes

API

Add start(EngineManifest) and stop(EngineManifest) to DebeziumConnectorRegistry interface

Engine Producers

(all 7 connectors: Postgres, MySQL, MariaDb, SqlServer, Oracle, MongoDB, Db2)

• Add ConcurrentHashMap<String, DebeziumRunner> to manage runner lifecycle
• start(): uses putIfAbsent for idempotency, cleans up map on failure
• stop(): uses remove atomically, logs error on shutdown failure without rethrowing

Configuration

Add quarkus.debezium.engine.autostart (default true) read in EngineProcessor

Test App

Add start/stop endpoints to EngineResource (single and per-id)
Add /engine/statuses endpoint for multi-engine status queries

Integration Tests

ManualStartSingleEngineIT: autostart=false, manual start, lifecycle events, restart, graceful shutdown when never started
ManualStartMultiEngineIT: single/all engine start/stop, state verification across engines

Close issue dbz#1662

[Naros]

I'm skeptical of not throwing the shutdown exception up to the caller here. Are there not use cases where it could be good for the caller here to know the runner failed to stop and it shouldn't attempt to restart it?

[kmos]

agree

[Naros]

In addition, start/stop appear to be mostly copy-n-paste between the producers. I'd rather introduce some sort of abstract class here and pull this up for now to minimize the dupe.

[kmos]

Yes, as @Naros mentioned, you currently have some duplication that suggests an opportunity for refactoring. My recommendation would be to extract a RunnableDebeziumConnectorRegistry that receives the engine map and the Connector in its constructor. @yuanglili WDYT? Let me know if this approach sounds reasonable.

[yuanglili]

Hi @kmos and @Naros , thanks for the feedback! The RunnableDebeziumConnectorRegistry approach makes sense! I will replace the anonymous class with it.

[kmos]

@yuanglili Thanks. I’ve added some comments for us to discuss.

[kmos]

Yes, as @Naros mentioned, you currently have some duplication that suggests an opportunity for refactoring. My recommendation would be to extract a RunnableDebeziumConnectorRegistry that receives the engine map and the Connector in its constructor. @yuanglili WDYT? Let me know if this approach sounds reasonable.

[kmos]

I’m not convinced that using a return is the right approach here. I would rather throw an exception such as IllegalStateException or IllegalDebeziumStateException. The same strategy should be applied consistently across methods like start and stop. @Naros @yuanglili WDYT?

[yuanglili]

I Agreed, My original thinking was that a duplicate start() call might not be critical, but a clear exception gives the caller immediate feedback would be better.

I don't see an existing IllegalDebeziumStateException in codebase , I can use IllegalStateException at here first I remember debezium/dbz#1662 (comment) @jpechane mentioned some broader lifecycle control features, I can create a IllegalDebeziumStateException in a follow-up PR for future use.

[Naros]

That's fine as long as we don't loose sight of it @yuanglili

[kmos]

You can create the exception in this PR. As for the lifecycle control feature, it’s fine to handle it in a separate PR. It's necessary some work around it

[yuanglili]

I’ll make the updates to this PR later this week, and handle the lifecycle control feature in a separate PR.

[kmos]

agree

[yuanglili]

@Naros @kmos Thanks for the feedback! Addressed all comments in the latest commit:

• Extracted RunnableDebeziumConnectorRegistry to decrease duplication across all 7 producers
• start()/stop() now throw IllegalDebeziumStateException
• Added IllegalDebeziumStateException in this PR

While adding restart support, I realized the current design stores pre-built Debezium instances, but AsyncEmbeddedEngine is effectively one-shot, so the same instance cannot be reused after stop().

I switched to Supplier<Debezium> so each start() creates a fresh engine. The awkward part is that the constructor calls supplier.get() upfront just so engines() has something to iterate over, and start() creates another instance anyway.

So I thinkengines()is currently doing two different jobs: describing registered engines and exposing live runtime instances. I'm considering adding manifests() to the interface so the recorder can iterate over registered engine IDs at startup without needing pre-built instances, and having engines() only return instances that have actually been started.

Would that direction make sense, or is there a simpler way you’d recommend?

[kmos]

@yuanglili can you fix the CI errors?

[yuanglili]

@yuanglili can you fix the CI errors?

@kmos Sure, I’ll take care of it. Last week and this week are finals weeks for me, so I’m a bit busy, but I’ll have the CI issues fixed before the end of this week.I’ll also get Create example connector submitted

[yuanglili]

@kmos @Naros
The CI error is because CompatibleModeConnectorRecorder.java needs to support the new methods, including start() and stop().
I also added manifests() to more clearly separate what’s registered from what’s actually running, so engines() now returns only live instances.

Summary of changes after the last review:
• Extracted RunnableDebeziumConnectorRegistry to reduce duplication across all 7 producers
• Updated start() / stop() to throw IllegalDebeziumStateException
• Added IllegalDebeziumStateException
• Added manifests() to the DebeziumConnectorRegistry interface
• Implemented the full lifecycle (start / stop / manifests) in CompatibleModeConnectorRecorder

Happy to iterate if you think there’s a better approach.

[kmos]

@kmos @Naros The CI error is because CompatibleModeConnectorRecorder.java needs to support the new methods, including start() and stop(). I also added manifests() to more clearly separate what’s registered from what’s actually running, so engines() now returns only live instances.

Summary of changes after the last review: • Extracted RunnableDebeziumConnectorRegistry to reduce duplication across all 7 producers • Updated start() / stop() to throw IllegalDebeziumStateException • Added IllegalDebeziumStateException • Added manifests() to the DebeziumConnectorRegistry interface • Implemented the full lifecycle (start / stop / manifests) in CompatibleModeConnectorRecorder

Happy to iterate if you think there’s a better approach.

Thanks @yuanglili. Let's give a run to the CI and a review from @Naros . I have only a doubt but overall it's a great work! 🚀

[kmos]

@Naros WDYT? I have a bias for RuntimeException but I am not sure if in this case should be an unchecked exception

[Naros]

@yuanglili @kmos I don't have an issue with throwing an unchecked exception in this context, as long as the methods that can throw it document the possible unchecked exceptions. This keeps in line with other Java APIs that throw IllegalStateException.

My concern here is whether we need a specialized exception class? Why can we not merely rely on the existing DebeziumException?

[kmos]

I don't have any concern to rely to DebeziumException

[yuanglili]

I'll switch to using DebeziumException.

[Naros]

Just a few inline comments.

[Naros]

Hi @yuanglili, can you rebase on the latest main?

[yuanglili]

Hi @yuanglili, can you rebase on the latest main?

Done, rebased on the latest main.

[yuanglili]

I noticed the Oracle CI failure, looking into it now. I usually skip Oracle tests locally on my Mac since
they're a pain to run on Apple Silicon, so didn't catch this until after the rebase. Sorry for the late catch.

[yuanglili]

@kmos @Naros
The root cause was thread.join() in DebeziumRunner.shutdown(). On Oracle, LogMiner teardown can take several seconds, causing Application.close() in lifecycle tests to hang and pushing DebeziumServerTest over the shared 100s timeout. I added a bounded 30s wait for the join.

While debugging, I also found two lifecycle issues:
• On putIfAbsent race-loss or runner.start() failure, the created Debezium instance was never closed, which could leak resources. I added closeQuietly in both paths.
• If runner.start() throws in CompatibleModeConnectorRecorder.start(), the runner stayed in the map and future starts would fail with “already running”. I added cleanup on failure.

[yuanglili]

The failed test passed on my local laptop. The problem might be: registry.stop() already returned, but SQL Server may still be cleaning up CDC cursors / JDBC connections in the background. The immediate next start hits that unfinished cleanup and fails, surfacing as 500 instead of 200.

Fix: in ManualStartSingleEngineIT.shouldRestartEngine, the test now retries the start call after stop until it succeeds (instead of expecting it to work on the first try).

[yuanglili]

@kmos Hi, I found the issue！

Failures down to one issue in QuarkusHeartbeatEmitter L55, engines().getFirst() throws NoSuchElementException when engine is stopping and list is empty, connector dies, all subsequent tests time out waiting for POLLING.

This wasn't hit before because a recent PR changed config phase to RUN_TIME with lazy resolve, which made tests run longer and exposed the race.

I can open another PR add an empty-list guard in emit() and rebase this PR on top. WDYT？

[kmos]

@kmos Hi, I found the issue！

Failures down to one issue in QuarkusHeartbeatEmitter L55, engines().getFirst() throws NoSuchElementException when engine is stopping and list is empty, connector dies, all subsequent tests time out waiting for POLLING.

This wasn't hit before because a recent PR changed config phase to RUN_TIME with lazy resolve, which made tests run longer and exposed the race.

I can open another PR add an empty-list guard in emit() and rebase this PR on top. WDYT？

@yuanglili, Can I give you a suggestion? in the emit try this:

        this.registries
                .stream()
                .flatMap(registry -> registry.engines().stream())
                .filter(engine -> engine.manifest().equals(context().manifest()))
                .map(engine ->  Map.entry(engine, heartbeat.select(Engine.Literal.of(engine.manifest().id()))))
                .forEach(entry -> entry.getValue().fire(
                        new DebeziumHeartbeat(
                                entry.getKey().connector(),
                                entry.getKey().status(),
                                partition,
                                offset.getOffset())));

and in the QuarkusHeartbeatEmitterTest add in the generated mock the manifest new EngineManifest("testing");

Let me know WDYT. About this PR or another one, it's perfectly fine for me

[yuanglili]

@kmos Hi, I found the issue！
Failures down to one issue in QuarkusHeartbeatEmitter L55, engines().getFirst() throws NoSuchElementException when engine is stopping and list is empty, connector dies, all subsequent tests time out waiting for POLLING.
This wasn't hit before because a recent PR changed config phase to RUN_TIME with lazy resolve, which made tests run longer and exposed the race.
I can open another PR add an empty-list guard in emit() and rebase this PR on top. WDYT？

@yuanglili, Can I give you a suggestion? in the emit try this:

        this.registries
                .stream()
                .flatMap(registry -> registry.engines().stream())
                .filter(engine -> engine.manifest().equals(context().manifest()))
                .map(engine ->  Map.entry(engine, heartbeat.select(Engine.Literal.of(engine.manifest().id()))))
                .forEach(entry -> entry.getValue().fire(
                        new DebeziumHeartbeat(
                                entry.getKey().connector(),
                                entry.getKey().status(),
                                partition,
                                offset.getOffset())));

and in the QuarkusHeartbeatEmitterTest add in the generated mock the manifest new EngineManifest("testing");

Let me know WDYT. About this PR or another one, it's perfectly fine for me

Sounds good, this approach is cleaner and safer than using getFirst(). I’ll open a separate PR for this so it’s easier to review and track independently, then rebase this PR on top of it.

[kmos]

@kmos Hi, I found the issue！
Failures down to one issue in QuarkusHeartbeatEmitter L55, engines().getFirst() throws NoSuchElementException when engine is stopping and list is empty, connector dies, all subsequent tests time out waiting for POLLING.
This wasn't hit before because a recent PR changed config phase to RUN_TIME with lazy resolve, which made tests run longer and exposed the race.
I can open another PR add an empty-list guard in emit() and rebase this PR on top. WDYT？

@yuanglili, Can I give you a suggestion? in the emit try this:

        this.registries
                .stream()
                .flatMap(registry -> registry.engines().stream())
                .filter(engine -> engine.manifest().equals(context().manifest()))
                .map(engine ->  Map.entry(engine, heartbeat.select(Engine.Literal.of(engine.manifest().id()))))
                .forEach(entry -> entry.getValue().fire(
                        new DebeziumHeartbeat(
                                entry.getKey().connector(),
                                entry.getKey().status(),
                                partition,
                                offset.getOffset())));

and in the QuarkusHeartbeatEmitterTest add in the generated mock the manifest new EngineManifest("testing");
Let me know WDYT. About this PR or another one, it's perfectly fine for me

Sounds good, this approach is cleaner and safer than using getFirst(). I’ll open a separate PR for this so it’s easier to review and track independently, then rebase this PR on top of it.

to be super-precise @yuanglili , it's also necesssary to address the DefaultEngine annotation:

        this.registries
                .stream()
                .flatMap(registry -> registry.engines().stream())
                .filter(engine -> engine.manifest().equals(context().manifest()))
                .map(engine -> Map.entry(engine, DefaultEngine.Literal.selectDefault(heartbeat.select(Engine.Literal.of(engine.manifest().id())), engine.manifest())))
                .forEach(entry -> entry.getValue().fire(
                        new DebeziumHeartbeat(
                                entry.getKey().connector(),
                                entry.getKey().status(),
                                partition,
                                offset.getOffset())));

[kmos]

@yuanglili I have opened a new GH issue about the heartbeat debezium/dbz#2006 . Please let me know if you have cycles to open a PR

[yuanglili]

@yuanglili I have opened a new GH issue about the heartbeat debezium/dbz#2006 . Please let me know if you have cycles to open a PR

@kmos PR is up : #56. Sorry for the delay. Once that PR is approved and merged, I’ll rebase this PR on top of it and update it right away.

[kmos]

@yuanglili I think there is still something failing for sqlserver

[yuanglili]

@yuanglili I think there is still something failing for sqlserver

I noticed the failing, it passes locall, probably CI is slower, so the SQL Server engine dies on its own before the test's /engine/stop runs. When that happens the engine is already closed, but the runners map still thinks it's alive → next stop calls close() again → IllegalStateException → 500.

I Added a catch (IllegalStateException) in DebeziumRunner.shutdown() to see if this is it. If CI passes, that's the problem.