Helper Utilities

The pydbzengine library provides a set of helper functions inside the Utils class (located in pydbzengine.helper) to help manage the engine's lifecycle, run the engine asynchronously, or run until snapshots complete.

Methods Reference

Utils.run_engine_async

Runs the given engine's run method in a separate background thread with a maximum duration (timeout). If the engine exceeds the timeout limit, it logs a warning and calls engine.close().

@staticmethod
def run_engine_async(
    engine: DebeziumEngine, timeout_sec: int = 22, blocking: bool = True
)

• engine: The DebeziumEngine (or DebeziumJsonEngine) instance to execute.
• timeout_sec: The maximum execution time in seconds. Defaults to 22.
• blocking: If True, blocks the calling thread (e.g. main thread) using join() until the engine completes or times out. Defaults to True.

Example Usage

from pydbzengine import DebeziumEngine
from pydbzengine.helper import Utils

# Run the engine for exactly 60 seconds and then stop
engine = DebeziumEngine(properties=props, handler=handler)
Utils.run_engine_async(engine=engine, timeout_sec=60)

---

Utils.run_engine_until_snapshot

Runs the Debezium engine in a daemon thread and monitors logging outputs. Once the log message "Snapshot completed" is detected, the engine is shut down gracefully. This is extremely useful for loading initial tables/snapshots when using snapshot.mode = initial_only (or equivalent Debezium snapshot modes) in batched ingestion jobs.

@staticmethod
def run_engine_until_snapshot(engine: DebeziumEngine, poll_interval_sec: int = 1)

• engine: The DebeziumEngine (or DebeziumJsonEngine) instance to execute.
• poll_interval_sec: The frequency (in seconds) to check log handlers for snapshot completion. Defaults to 1.

Example Usage

from pydbzengine import DebeziumEngine
from pydbzengine.helper import Utils

# Config properties include: "snapshot.mode": "initial_only"
engine = DebeziumEngine(properties=props, handler=handler)
Utils.run_engine_until_snapshot(engine=engine)