Implementation:Risingwavelabs Risingwave CdcEngine Interface

Metadata

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

Overview

CdcEngine is a core interface that defines the contract for a Change Data Capture (CDC) engine in the RisingWave connector framework. By extending Runnable, it is designed to execute within a thread and produce a stream of change events that are published to a BlockingQueue.

The interface provides methods for retrieving the engine's unique identifier, stopping the engine, and accessing the output channel through which GetEventStreamResponse protobuf messages are delivered. Different database connectors (MySQL, PostgreSQL, MongoDB, etc.) implement this interface to produce change events in a uniform way.

Code Reference

Source Location

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

Signature

public interface CdcEngine extends Runnable {
    long getId();

    void stop() throws Exception;

    BlockingQueue<ConnectorServiceProto.GetEventStreamResponse> getOutputChannel();
}

Imports

import com.risingwave.proto.ConnectorServiceProto;
import java.util.concurrent.BlockingQueue;

I/O Contract

getId

Direction	Type	Description
Output	long	The unique identifier of this CDC engine instance

stop

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

May throw Exception if an error occurs during shutdown.

getOutputChannel

Direction	Type	Description
Output	BlockingQueue<ConnectorServiceProto.GetEventStreamResponse>	The blocking queue through which CDC events are delivered

Consumers can poll or take events from this queue to receive change data capture events from the underlying database.

run (inherited from Runnable)

When executed (typically in a thread), the engine connects to the source database, monitors for changes, and publishes GetEventStreamResponse messages to the output channel.

Usage Examples

// Typical usage within a CdcEngineRunner
CdcEngine engine = createEngine(sourceId, config);

// Start the engine in a dedicated thread
Thread engineThread = new Thread(engine);
engineThread.start();

// Consume events from the output channel
BlockingQueue<ConnectorServiceProto.GetEventStreamResponse> channel = engine.getOutputChannel();
while (running) {
    ConnectorServiceProto.GetEventStreamResponse event = channel.take();
    // Forward event to the gRPC response stream
    responseObserver.onNext(event);
}

// Stop the engine when done
engine.stop();

Related Pages

• CdcEngineRunner Interface - Manages the lifecycle of a CdcEngine instance
• SourceHandler Interface - Source handler that uses CdcEngine to stream events
• SourceTypeE - Enum of supported CDC source database types
• DbzCdcEngineRunner Start - Debezium-based implementation of the CDC engine runner