Package org.apache.flink.cdc.debezium

Class 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.9/development/engine.html#engine-properties

    See Also:
    Serialized Form

    • 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

    • 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()