Implementation:Risingwavelabs Risingwave ReplicationStream

Property	Value
Component	CDC Source - Debezium PostgreSQL Connector
Language	Java
Lines	144
License	Apache 2.0 (RisingWave Labs + Debezium Authors)
Repository	risingwavelabs/risingwave

Overview

ReplicationStream is a patched version of the Debezium ReplicationStream interface that defines the contract for consuming messages from a PostgreSQL logical replication stream. This interface is used by RisingWave's PostgreSQL CDC source to read change data capture events over a replication connection.

The interface extends AutoCloseable and provides methods for blocking and non-blocking reads of replication messages, LSN (Log Sequence Number) management, and keep-alive signal handling. It also includes a nested ReplicationMessageProcessor functional interface that serves as the callback mechanism for processing individual replication messages.

This file is placed in the io.debezium.connector.postgresql.connection package to override the upstream Debezium interface, enabling RisingWave-specific extensions to the replication stream contract, including additional methods on the ReplicationMessageProcessor for schema change event dispatching.

Code Reference

Source Location

java/connector-node/risingwave-source-cdc/src/main/java/io/debezium/connector/postgresql/connection/ReplicationStream.java

View on GitHub

Signature

public interface ReplicationStream extends AutoCloseable

Nested Interface: ReplicationMessageProcessor

@FunctionalInterface
interface ReplicationMessageProcessor {
    // Core message processing callback
    void process(ReplicationMessage message) throws SQLException, InterruptedException;

    // Returns the Debezium event dispatcher for schema change events
    default EventDispatcher<PostgresPartition, TableId> getEventDispatcher();

    // Returns the PostgreSQL partition context
    default PostgresPartition getPartition();

    // Returns the current offset context
    default OffsetContext getOffsetContext();

    // Whether to include schema change events
    default boolean includeSchemaChange();
}

Key Methods

// Blocking read: waits for a replication message and passes it to the processor
void read(ReplicationMessageProcessor processor) throws SQLException, InterruptedException;

// Non-blocking read: returns true if a message was available and processed
boolean readPending(ReplicationMessageProcessor processor) throws SQLException, InterruptedException;

// Informs the server about the latest successfully processed WAL position
void flushLsn(Lsn lsn) throws SQLException;

// Returns the last LSN received during a read operation
Lsn lastReceivedLsn();

// Returns the LSN from which streaming started
Lsn startLsn();

// Starts a background keep-alive thread to maintain the replication slot
void startKeepAlive(ExecutorService service);

// Stops the background keep-alive thread
void stopKeepAlive();

// Closes the replication stream
@Override
void close() throws Exception;

Imports

import io.debezium.connector.postgresql.PostgresPartition;
import io.debezium.pipeline.EventDispatcher;
import io.debezium.pipeline.spi.OffsetContext;
import io.debezium.relational.TableId;
import java.sql.SQLException;
import java.util.concurrent.ExecutorService;
import org.postgresql.replication.PGReplicationStream;

I/O Contract

Input

• ReplicationMessageProcessor: A callback that receives ReplicationMessage objects for processing. Default methods provide access to the event dispatcher, partition, offset context, and schema change inclusion flag.
• Lsn: A Log Sequence Number value passed to flushLsn() to acknowledge processed WAL positions.
• ExecutorService: Provided to startKeepAlive() for running the background keep-alive thread.

Output

• read(): Blocks until a message is received; updates lastReceivedLsn().
• readPending(): Returns true if a message was available and processed, false otherwise.
• lastReceivedLsn(): Returns the most recent LSN received during read operations.
• startLsn(): Returns the LSN from which streaming was initiated.

Side Effects

• WAL acknowledgment: flushLsn() sends an acknowledgment to the PostgreSQL server, allowing it to discard older WAL segments.
• Slot keep-alive: startKeepAlive()/stopKeepAlive() manage a background thread that prevents the replication slot from being dropped due to inactivity.

Usage Examples

Reading Messages from a Replication Stream

ReplicationStream stream = // obtained from ReplicationConnection.startStreaming()

// Blocking read
stream.read(message -> {
    // Process the replication message
    handleChange(message);
});

// Non-blocking read
boolean hasMessage = stream.readPending(message -> {
    handleChange(message);
});

// Acknowledge processed position
stream.flushLsn(stream.lastReceivedLsn());

Managing Keep-Alive

ExecutorService keepAliveService = Executors.newSingleThreadExecutor();
stream.startKeepAlive(keepAliveService);
// ... perform metadata queries or other operations ...
stream.stopKeepAlive();

Related Pages

• DbzCdcEngineRunner Start - CDC engine runner that manages PostgreSQL CDC sources
• ConfigurableOffsetBackingStore - Offset backing store for CDC offset management
• JniDbzSourceHandler RunJniDbzSourceThread - JNI bridge for Debezium source threads