com.ververica.cdc.debezium

Class DebeziumSourceFunction<T>

java.lang.Object

org.apache.flink.api.common.functions.AbstractRichFunction

org.apache.flink.streaming.api.functions.source.RichSourceFunction<T>

com.ververica.cdc.debezium.DebeziumSourceFunction<T>

All Implemented Interfaces:

Serializable, org.apache.flink.api.common.functions.Function, org.apache.flink.api.common.functions.RichFunction, org.apache.flink.api.common.state.CheckpointListener, org.apache.flink.api.java.typeutils.ResultTypeQueryable<T>, org.apache.flink.streaming.api.checkpoint.CheckpointedFunction, org.apache.flink.streaming.api.functions.source.SourceFunction<T>

@PublicEvolving
public class DebeziumSourceFunction<T>
extends org.apache.flink.streaming.api.functions.source.RichSourceFunction<T>
implements org.apache.flink.streaming.api.checkpoint.CheckpointedFunction, org.apache.flink.api.common.state.CheckpointListener, org.apache.flink.api.java.typeutils.ResultTypeQueryable<T>

The DebeziumSourceFunction is a streaming data source that pulls captured change data from databases into Flink.

There are two workers during the runtime. One worker periodically pulls records from the database and pushes the records into the Handover. The other worker consumes the records from the Handover and convert the records to the data in Flink style. The reason why don't use one workers is because debezium has different behaviours in snapshot phase and streaming phase.

Here we use the Handover as the buffer to submit data from the producer to the consumer. Because the two threads don't communicate to each other directly, the error reporting also relies on Handover. When the engine gets errors, the engine uses the DebeziumEngine.CompletionCallback to report errors to the Handover and wakes up the consumer to check the error. However, the source function just closes the engine and wakes up the producer if the error is from the Flink side.

If the execution is canceled or finish(only snapshot phase), the exit logic is as same as the logic in the error reporting.

The source function participates in checkpointing and guarantees that no data is lost during a failure, and that the computation processes elements "exactly once".

Note: currently, the source function can't run in multiple parallel instances.

Please refer to Debezium's documentation for the available configuration properties: https://debezium.io/documentation/reference/1.5/development/engine.html#engine-properties

See Also:

Serialized Form

Nested Class Summary

Nested classes/interfaces inherited from interface org.apache.flink.streaming.api.functions.source.SourceFunction

org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext<T>

Field Summary

Fields

Modifier and Type	Field and Description
static String	HISTORY_RECORDS_STATE_NAME State name of the consumer's history records state.
static String	LEGACY_IMPLEMENTATION_KEY The configuration represents the Debezium MySQL Connector uses the legacy implementation or not.
static String	LEGACY_IMPLEMENTATION_VALUE The configuration value represents legacy implementation.
protected static org.slf4j.Logger	LOG
static int	MAX_NUM_PENDING_CHECKPOINTS The maximum number of pending non-committed checkpoints to track, to avoid memory leaks.
static String	OFFSETS_STATE_NAME State name of the consumer's partition offset states.

Constructor Summary

Constructors

Constructor and Description
DebeziumSourceFunction(DebeziumDeserializationSchema<T> deserializer, Properties properties, DebeziumOffset specificOffset, Validator validator)

Method Summary

Modifier and Type	Method and Description
void	cancel()
void	close()
boolean	getDebeziumStarted()
String	getEngineInstanceName()
org.apache.commons.collections.map.LinkedMap	getPendingOffsetsToCommit()
org.apache.flink.api.common.typeinfo.TypeInformation<T>	getProducedType()
void	initializeState(org.apache.flink.runtime.state.FunctionInitializationContext context)
void	notifyCheckpointComplete(long checkpointId)
void	open(org.apache.flink.configuration.Configuration parameters)
void	run(org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext<T> sourceContext)
void	snapshotState(org.apache.flink.runtime.state.FunctionSnapshotContext functionSnapshotContext)

Methods inherited from class org.apache.flink.api.common.functions.AbstractRichFunction

getIterationRuntimeContext, getRuntimeContext, setRuntimeContext

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.apache.flink.api.common.state.CheckpointListener

notifyCheckpointAborted

Field Detail

LOG

protected static final org.slf4j.Logger LOG

OFFSETS_STATE_NAME

public static final String OFFSETS_STATE_NAME

State name of the consumer's partition offset states.

See Also:

Constant Field Values

HISTORY_RECORDS_STATE_NAME

public static final String HISTORY_RECORDS_STATE_NAME

State name of the consumer's history records state.

See Also:

Constant Field Values

MAX_NUM_PENDING_CHECKPOINTS

public static final int MAX_NUM_PENDING_CHECKPOINTS

The maximum number of pending non-committed checkpoints to track, to avoid memory leaks.

See Also:

Constant Field Values

LEGACY_IMPLEMENTATION_KEY

public static final String LEGACY_IMPLEMENTATION_KEY

The configuration represents the Debezium MySQL Connector uses the legacy implementation or not.

See Also:

Constant Field Values

LEGACY_IMPLEMENTATION_VALUE

public static final String LEGACY_IMPLEMENTATION_VALUE

The configuration value represents legacy implementation.

See Also:

Constant Field Values

Constructor Detail

DebeziumSourceFunction

public DebeziumSourceFunction(DebeziumDeserializationSchema<T> deserializer,
                              Properties properties,
                              @Nullable
                              DebeziumOffset specificOffset,
                              Validator validator)

Method Detail

open

public void open(org.apache.flink.configuration.Configuration parameters)
          throws Exception

Specified by:

open in interface org.apache.flink.api.common.functions.RichFunction

Overrides:

open in class org.apache.flink.api.common.functions.AbstractRichFunction

Throws:

Exception

initializeState

public void initializeState(org.apache.flink.runtime.state.FunctionInitializationContext context)
                     throws Exception

Specified by:

initializeState in interface org.apache.flink.streaming.api.checkpoint.CheckpointedFunction

Throws:

Exception

snapshotState

public void snapshotState(org.apache.flink.runtime.state.FunctionSnapshotContext functionSnapshotContext)
                   throws Exception

Specified by:

snapshotState in interface org.apache.flink.streaming.api.checkpoint.CheckpointedFunction

Throws:

Exception

run

public void run(org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext<T> sourceContext)
         throws Exception

Specified by:

run in interface org.apache.flink.streaming.api.functions.source.SourceFunction<T>

Throws:

Exception

notifyCheckpointComplete

public void notifyCheckpointComplete(long checkpointId)

Specified by:

notifyCheckpointComplete in interface org.apache.flink.api.common.state.CheckpointListener

cancel

public void cancel()

Specified by:

cancel in interface org.apache.flink.streaming.api.functions.source.SourceFunction<T>

close

public void close()
           throws Exception

Specified by:

close in interface org.apache.flink.api.common.functions.RichFunction

Overrides:

close in class org.apache.flink.api.common.functions.AbstractRichFunction

Throws:

Exception

getProducedType

public org.apache.flink.api.common.typeinfo.TypeInformation<T> getProducedType()

Specified by:

getProducedType in interface org.apache.flink.api.java.typeutils.ResultTypeQueryable<T>

getPendingOffsetsToCommit

@VisibleForTesting
public org.apache.commons.collections.map.LinkedMap getPendingOffsetsToCommit()

getDebeziumStarted

@VisibleForTesting
public boolean getDebeziumStarted()

getEngineInstanceName

@VisibleForTesting
public String getEngineInstanceName()