org.springframework.cloud.fn.common.cdc

Class EmbeddedEngine

java.lang.Object

org.springframework.cloud.fn.common.cdc.EmbeddedEngine

All Implemented Interfaces:

io.debezium.engine.DebeziumEngine<org.apache.kafka.connect.source.SourceRecord>, Closeable, AutoCloseable, Runnable

@ThreadSafe
public final class EmbeddedEngine
extends Object
implements io.debezium.engine.DebeziumEngine<org.apache.kafka.connect.source.SourceRecord>

A mechanism for running a single Kafka Connect SourceConnector within an application's process. An embedded connector is entirely standalone and only talks with the source system; no Kafka, Kafka Connect, or Zookeeper processes are needed. Applications using an embedded connector simply set one up and supply a consumer function to which the connector will pass all SourceRecords containing database change events.

With an embedded connector, the application that runs the connector assumes all responsibility for fault tolerance, scalability, and durability. Additionally, applications must specify how the connector can store its relational database schema history and offsets. By default, this information will be stored in memory and will thus be lost upon application restart.

Embedded connectors are designed to be submitted to an Executor or ExecutorService for execution by a single thread, and a running connector can be stopped either by calling stop() from another thread or by interrupting the running thread (e.g., as is the case with ExecutorService.shutdownNow()).

Author:

Randall Hauch

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static interface	EmbeddedEngine.Builder Deprecated.
static class	EmbeddedEngine.BuilderImpl
static interface	EmbeddedEngine.ChangeConsumer Deprecated.
static interface	EmbeddedEngine.CompletionCallback Deprecated.
static class	EmbeddedEngine.CompletionResult A callback function to be notified when the connector completes.
static interface	EmbeddedEngine.ConnectorCallback Deprecated.
protected static class	EmbeddedEngine.EmbeddedConfig
static interface	EmbeddedEngine.RecordCommitter Deprecated.
protected class	EmbeddedEngine.SourceRecordOffsets

Nested classes/interfaces inherited from interface io.debezium.engine.DebeziumEngine

io.debezium.engine.DebeziumEngine.BuilderFactory, io.debezium.engine.DebeziumEngine.Offsets

Field Summary

Fields

Modifier and Type	Field and Description
protected static io.debezium.config.Field.Set	ALL_FIELDS The array of all exposed fields.
static io.debezium.config.Field	CONNECTOR_CLASS A required field for an embedded connector that specifies the name of the normal Debezium connector's Java class.
static io.debezium.config.Field.Set	CONNECTOR_FIELDS The array of fields that are required by each connectors.
static io.debezium.config.Field	ENGINE_NAME A required field for an embedded connector that specifies the unique name for the connector instance.
protected static io.debezium.config.Field	INTERNAL_KEY_CONVERTER_CLASS
protected static io.debezium.config.Field	INTERNAL_VALUE_CONVERTER_CLASS
static io.debezium.config.Field	OFFSET_COMMIT_POLICY
static io.debezium.config.Field	OFFSET_COMMIT_TIMEOUT_MS An optional advanced field that specifies the maximum amount of time that the embedded connector should wait for an offset commit to complete.
static io.debezium.config.Field	OFFSET_FLUSH_INTERVAL_MS An optional advanced field that specifies the maximum amount of time that the embedded connector should wait for an offset commit to complete.
static io.debezium.config.Field	OFFSET_STORAGE An optional field that specifies the name of the class that implements the OffsetBackingStore interface, and that will be used to store offsets recorded by the connector.
static io.debezium.config.Field	OFFSET_STORAGE_FILE_FILENAME An optional field that specifies the file location for the FileOffsetBackingStore.
static io.debezium.config.Field	OFFSET_STORAGE_KAFKA_PARTITIONS An optional field that specifies the number of partitions for the KafkaOffsetBackingStore.
static io.debezium.config.Field	OFFSET_STORAGE_KAFKA_REPLICATION_FACTOR An optional field that specifies the replication factor for the KafkaOffsetBackingStore.
static io.debezium.config.Field	OFFSET_STORAGE_KAFKA_TOPIC An optional field that specifies the topic name for the KafkaOffsetBackingStore.
static io.debezium.config.Field	TRANSFORMS A list of SMTs to be applied on the messages generated by the engine.

Fields inherited from interface io.debezium.engine.DebeziumEngine

OFFSET_FLUSH_INTERVAL_MS_PROP

Method Summary

Modifier and Type	Method and Description
boolean	await(long timeout, TimeUnit unit) Wait for the connector to complete processing.
protected EmbeddedEngine.RecordCommitter	buildRecordCommitter(org.apache.kafka.connect.storage.OffsetStorageWriter offsetWriter, org.apache.kafka.connect.source.SourceTask task, Duration commitTimeout) Creates a new RecordCommitter that is responsible for informing the engine about the updates to the given batch
void	close()
protected void	commitOffsets(org.apache.kafka.connect.storage.OffsetStorageWriter offsetWriter, Duration commitTimeout, org.apache.kafka.connect.source.SourceTask task) Flush offsets to storage.
protected void	completedFlush(Throwable error, Void result)
static EmbeddedEngine.Builder	create() Deprecated.
boolean	isRunning() Determine if this embedded connector is currently running.
protected void	maybeFlush(org.apache.kafka.connect.storage.OffsetStorageWriter offsetWriter, io.debezium.engine.spi.OffsetCommitPolicy policy, Duration commitTimeout, org.apache.kafka.connect.source.SourceTask task) Determine if we should flush offsets to storage, and if so then attempt to flush offsets.
void	run() Run this embedded connector and deliver database changes to the registered Consumer.
void	runWithTask(Consumer<org.apache.kafka.connect.source.SourceTask> consumer)
boolean	stop() Stop the execution of this embedded connector.
String	toString()

Methods inherited from class java.lang.Object

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

Methods inherited from interface io.debezium.engine.DebeziumEngine

create, create, create, create

Field Detail

ENGINE_NAME

public static final io.debezium.config.Field ENGINE_NAME

A required field for an embedded connector that specifies the unique name for the connector instance.

CONNECTOR_CLASS

public static final io.debezium.config.Field CONNECTOR_CLASS

A required field for an embedded connector that specifies the name of the normal Debezium connector's Java class.

OFFSET_STORAGE

public static final io.debezium.config.Field OFFSET_STORAGE

An optional field that specifies the name of the class that implements the OffsetBackingStore interface, and that will be used to store offsets recorded by the connector.

OFFSET_STORAGE_FILE_FILENAME

public static final io.debezium.config.Field OFFSET_STORAGE_FILE_FILENAME

An optional field that specifies the file location for the FileOffsetBackingStore.

See Also:

OFFSET_STORAGE

OFFSET_STORAGE_KAFKA_TOPIC

public static final io.debezium.config.Field OFFSET_STORAGE_KAFKA_TOPIC

An optional field that specifies the topic name for the KafkaOffsetBackingStore.

See Also:

OFFSET_STORAGE

OFFSET_STORAGE_KAFKA_PARTITIONS

public static final io.debezium.config.Field OFFSET_STORAGE_KAFKA_PARTITIONS

An optional field that specifies the number of partitions for the KafkaOffsetBackingStore.

See Also:

OFFSET_STORAGE

OFFSET_STORAGE_KAFKA_REPLICATION_FACTOR

public static final io.debezium.config.Field OFFSET_STORAGE_KAFKA_REPLICATION_FACTOR

An optional field that specifies the replication factor for the KafkaOffsetBackingStore.

See Also:

OFFSET_STORAGE

OFFSET_FLUSH_INTERVAL_MS

public static final io.debezium.config.Field OFFSET_FLUSH_INTERVAL_MS

An optional advanced field that specifies the maximum amount of time that the embedded connector should wait for an offset commit to complete.

OFFSET_COMMIT_TIMEOUT_MS

public static final io.debezium.config.Field OFFSET_COMMIT_TIMEOUT_MS

An optional advanced field that specifies the maximum amount of time that the embedded connector should wait for an offset commit to complete.

OFFSET_COMMIT_POLICY

public static final io.debezium.config.Field OFFSET_COMMIT_POLICY

INTERNAL_KEY_CONVERTER_CLASS

protected static final io.debezium.config.Field INTERNAL_KEY_CONVERTER_CLASS

INTERNAL_VALUE_CONVERTER_CLASS

protected static final io.debezium.config.Field INTERNAL_VALUE_CONVERTER_CLASS

TRANSFORMS

public static final io.debezium.config.Field TRANSFORMS

A list of SMTs to be applied on the messages generated by the engine.

CONNECTOR_FIELDS

public static final io.debezium.config.Field.Set CONNECTOR_FIELDS

The array of fields that are required by each connectors.

ALL_FIELDS

protected static final io.debezium.config.Field.Set ALL_FIELDS

The array of all exposed fields.

Method Detail

create

@Deprecated
public static EmbeddedEngine.Builder create()

Deprecated.

Obtain a new EmbeddedEngine.Builder instance that can be used to construct runnable EmbeddedEngine instances.

Returns:

the new builder; never null

isRunning

public boolean isRunning()

Determine if this embedded connector is currently running.

Returns:

true if running, or false otherwise

run

public void run()

Run this embedded connector and deliver database changes to the registered Consumer. This method blocks until the connector is stopped.

First, the method checks to see if this instance is currently running, and if so immediately returns.

If the configuration is valid, this method starts the connector and starts polling the connector for change events. All messages are delivered in batches to the Consumer registered with this embedded connector. The batch size, polling frequency, and other parameters are controlled via configuration settings. This continues until this connector is stopped.

Note that there are two ways to stop a connector running on a thread: calling stop() from another thread, or interrupting the thread (e.g., via ExecutorService.shutdownNow()).

This method can be called repeatedly as needed.

Specified by:

run in interface Runnable

buildRecordCommitter

protected EmbeddedEngine.RecordCommitter buildRecordCommitter(org.apache.kafka.connect.storage.OffsetStorageWriter offsetWriter,
                                                              org.apache.kafka.connect.source.SourceTask task,
                                                              Duration commitTimeout)

Creates a new RecordCommitter that is responsible for informing the engine about the updates to the given batch

Parameters:

offsetWriter - the offsetWriter current in use

task - the sourcetask

commitTimeout - the time in ms until a commit times out

Returns:

the new recordCommitter to be used for a given batch

maybeFlush

protected void maybeFlush(org.apache.kafka.connect.storage.OffsetStorageWriter offsetWriter,
                          io.debezium.engine.spi.OffsetCommitPolicy policy,
                          Duration commitTimeout,
                          org.apache.kafka.connect.source.SourceTask task)

Determine if we should flush offsets to storage, and if so then attempt to flush offsets.

Parameters:

offsetWriter - the offset storage writer; may not be null

policy - the offset commit policy; may not be null

commitTimeout - the timeout to wait for commit results

task - the task which produced the records for which the offsets have been committed

commitOffsets

protected void commitOffsets(org.apache.kafka.connect.storage.OffsetStorageWriter offsetWriter,
                             Duration commitTimeout,
                             org.apache.kafka.connect.source.SourceTask task)

Flush offsets to storage.

Parameters:

offsetWriter - the offset storage writer; may not be null

commitTimeout - the timeout to wait for commit results

task - the task which produced the records for which the offsets have been committed

completedFlush

protected void completedFlush(Throwable error,
                              Void result)

stop

public boolean stop()

Stop the execution of this embedded connector. This method does not block until the connector is stopped; use await(long, TimeUnit) for this purpose.

Returns:

true if the connector was running and will eventually stop, or false if it was not running when this method is called

See Also:

await(long, TimeUnit)

close

public void close()
           throws IOException

Specified by:

close in interface Closeable

Specified by:

close in interface AutoCloseable

Throws:

IOException

await

public boolean await(long timeout,
                     TimeUnit unit)
              throws InterruptedException

Wait for the connector to complete processing. If the processor is not running, this method returns immediately; however, if the processor is stopped and restarted before this method is called, this method will return only when it completes the second time.

Parameters:

timeout - the maximum amount of time to wait before returning

unit - the unit of time; may not be null

Returns:

true if the connector completed within the timeout (or was not running), or false if it is still running when the timeout occurred

Throws:

InterruptedException - if this thread is interrupted while waiting for the completion of the connector

toString

public String toString()

Overrides:

toString in class Object

runWithTask

public void runWithTask(Consumer<org.apache.kafka.connect.source.SourceTask> consumer)