Package org.apache.flink.cdc.debezium.internal

Class FlinkDatabaseHistory

java.lang.Object

io.debezium.relational.history.AbstractDatabaseHistory

org.apache.flink.cdc.debezium.internal.FlinkDatabaseHistory

All Implemented Interfaces:

io.debezium.relational.history.DatabaseHistory

public class FlinkDatabaseHistory
extends io.debezium.relational.history.AbstractDatabaseHistory

Inspired from MemoryDatabaseHistory but we will store the HistoryRecords in Flink's state for persistence.

Note: This is not a clean solution because we depends on a global variable and all the history records will be stored in state (grow infinitely). We may need to come up with a FileSystemDatabaseHistory in the future to store history in HDFS.

Field Summary

Fields

Modifier and Type	Field	Description
static String	DATABASE_HISTORY_INSTANCE_NAME

Fields inherited from class io.debezium.relational.history.AbstractDatabaseHistory

config, INTERNAL_PREFER_DDL, logger

Fields inherited from interface io.debezium.relational.history.DatabaseHistory

CONFIGURATION_FIELD_PREFIX_STRING, DDL_FILTER, NAME, SKIP_UNPARSEABLE_DDL_STATEMENTS, STORE_ONLY_CAPTURED_TABLES_DDL, STORE_ONLY_MONITORED_TABLES_DDL

Constructor Summary

Constructors

Constructor	Description
FlinkDatabaseHistory()

Method Summary

Modifier and Type	Method	Description
void	configure(io.debezium.config.Configuration config, io.debezium.relational.history.HistoryRecordComparator comparator, io.debezium.relational.history.DatabaseHistoryListener listener, boolean useCatalogBeforeSchema)
boolean	exists()
static boolean	isCompatible(Collection<SchemaRecord> records)	Determine whether the FlinkDatabaseHistory is compatible with the specified state.
protected void	recoverRecords(java.util.function.Consumer<io.debezium.relational.history.HistoryRecord> records)
void	stop()
boolean	storageExists()
protected void	storeRecord(io.debezium.relational.history.HistoryRecord record)
String	toString()

Methods inherited from class io.debezium.relational.history.AbstractDatabaseHistory

initializeStorage, record, record, recover, skipUnparseableDdlStatements, start, storeOnlyCapturedTables

Methods inherited from class java.lang.Object

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

Methods inherited from interface io.debezium.relational.history.DatabaseHistory

recover, recover

Field Detail

DATABASE_HISTORY_INSTANCE_NAME

public static final String DATABASE_HISTORY_INSTANCE_NAME

See Also:

Constant Field Values

Constructor Detail

FlinkDatabaseHistory

public FlinkDatabaseHistory()

Method Detail

configure

public void configure(io.debezium.config.Configuration config,
                      io.debezium.relational.history.HistoryRecordComparator comparator,
                      io.debezium.relational.history.DatabaseHistoryListener listener,
                      boolean useCatalogBeforeSchema)

Specified by:

configure in interface io.debezium.relational.history.DatabaseHistory

Overrides:

configure in class io.debezium.relational.history.AbstractDatabaseHistory

stop

public void stop()

Specified by:

stop in interface io.debezium.relational.history.DatabaseHistory

Overrides:

stop in class io.debezium.relational.history.AbstractDatabaseHistory

storeRecord

protected void storeRecord(io.debezium.relational.history.HistoryRecord record)
                    throws io.debezium.relational.history.DatabaseHistoryException

Specified by:

storeRecord in class io.debezium.relational.history.AbstractDatabaseHistory

Throws:

io.debezium.relational.history.DatabaseHistoryException

recoverRecords

protected void recoverRecords(java.util.function.Consumer<io.debezium.relational.history.HistoryRecord> records)

Specified by:

recoverRecords in class io.debezium.relational.history.AbstractDatabaseHistory

exists

public boolean exists()

storageExists

public boolean storageExists()

toString

public String toString()

Overrides:

toString in class Object

isCompatible

public static boolean isCompatible(Collection<SchemaRecord> records)

Determine whether the FlinkDatabaseHistory is compatible with the specified state.