Implementation:Risingwavelabs Risingwave CdcEngineRunner Interface

Metadata

Property	Value
File	java/connector-node/connector-api/src/main/java/com/risingwave/connector/api/source/CdcEngineRunner.java
Language	Java
Module	connector-api
Package	com.risingwave.connector.api.source
Classes	CdcEngineRunner (interface)
Lines	27

Overview

CdcEngineRunner is an interface that defines the lifecycle management contract for a CDC engine. It separates the concerns of engine execution management from the engine implementation itself, providing methods to start, stop, and query the running state of a CdcEngine, as well as to retrieve the underlying engine instance.

Implementations of this interface are responsible for managing the thread or execution context in which the CdcEngine runs, handling startup/shutdown sequencing, and monitoring the engine's health.

Code Reference

Source Location

java/connector-node/connector-api/src/main/java/com/risingwave/connector/api/source/CdcEngineRunner.java

Signature

public interface CdcEngineRunner {
    boolean start() throws Exception;

    void stop() throws Exception;

    CdcEngine getEngine();

    boolean isRunning();
}

Imports

No external imports beyond the package-local CdcEngine reference.

I/O Contract

start

Direction	Type	Description
Output	boolean	Returns true if the engine was started successfully, false otherwise

May throw Exception if an error occurs during startup. Implementations typically submit the CdcEngine (which is a Runnable) to an executor service.

stop

Direction	Type	Description
Output	None	Stops the running CDC engine and releases associated resources

May throw Exception if an error occurs during shutdown.

getEngine

Direction	Type	Description
Output	CdcEngine	The underlying CDC engine managed by this runner

Provides access to the engine instance, allowing callers to retrieve the output channel or engine ID.

isRunning

Direction	Type	Description
Output	boolean	true if the engine is currently running, false otherwise

Usage Examples

// Create and start a CDC engine runner
CdcEngineRunner runner = createRunner(sourceConfig);

try {
    boolean started = runner.start();
    if (!started) {
        throw new RuntimeException("Failed to start CDC engine");
    }

    // Access the underlying engine for its output channel
    CdcEngine engine = runner.getEngine();
    BlockingQueue<ConnectorServiceProto.GetEventStreamResponse> output =
            engine.getOutputChannel();

    // Monitor engine status
    while (runner.isRunning()) {
        ConnectorServiceProto.GetEventStreamResponse event = output.poll(1, TimeUnit.SECONDS);
        if (event != null) {
            responseObserver.onNext(event);
        }
    }
} finally {
    runner.stop();
}

Related Pages

• CdcEngine Interface - The engine interface managed by this runner
• DbzCdcEngineRunner Start - Debezium-based concrete implementation of CdcEngineRunner
• SourceHandler Interface - Source handlers that use CdcEngineRunner to manage CDC streams
• SourceTypeE - Enum of supported source types that determine which engine runner to create