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

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment