io.debezium
debezium-embedded
diff --git a/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/EmbeddedEngine.java b/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/EmbeddedEngine.java
index 4f281416..3b896f66 100644
--- a/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/EmbeddedEngine.java
+++ b/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/EmbeddedEngine.java
@@ -63,19 +63,19 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * A mechanism for running a single Kafka Connect {@link 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 {@link Consumer consumer function} to which the
- * connector will pass all {@link SourceRecord}s containing database change events.
+ * A mechanism for running a single Kafka Connect {@link 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 {@link Consumer
+ * consumer function} to which the connector will pass all {@link SourceRecord}s 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.
+ * 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 {@link Executor} or {@link ExecutorService} for execution by a single
- * thread, and a running connector can be stopped either by calling {@link #stop()} from another thread or by interrupting
- * the running thread (e.g., as is the case with {@link ExecutorService#shutdownNow()}).
+ * Embedded connectors are designed to be submitted to an {@link Executor} or {@link ExecutorService} for execution by a
+ * single thread, and a running connector can be stopped either by calling {@link #stop()} from another thread or by
+ * interrupting the running thread (e.g., as is the case with {@link ExecutorService#shutdownNow()}).
*
* @author Randall Hauch
*/
@@ -87,14 +87,14 @@ public final class EmbeddedEngine implements DebeziumEngine {
*/
public static final Field ENGINE_NAME = Field.create("name")
.withDescription("Unique name for this connector instance.")
- .withValidation(Field::isRequired);
+ .required();
/**
* A required field for an embedded connector that specifies the name of the normal Debezium connector's Java class.
*/
public static final Field CONNECTOR_CLASS = Field.create("connector.class")
.withDescription("The Java class for the connector")
- .withValidation(Field::isRequired);
+ .required();
/**
* An optional field that specifies the name of the class that implements the {@link OffsetBackingStore} interface,
@@ -151,22 +151,22 @@ public final class EmbeddedEngine implements DebeziumEngine {
+ KafkaOffsetBackingStore.class.getName() + " class.");
/**
- * An optional advanced field that specifies the maximum amount of time that the embedded connector should wait
- * for an offset commit to complete.
+ * An optional advanced field that specifies the maximum amount of time that the embedded connector should wait for
+ * an offset commit to complete.
*/
public static final Field OFFSET_FLUSH_INTERVAL_MS = Field.create("offset.flush.interval.ms")
- .withDescription("Interval at which to try committing offsets. The default is 1 minute.")
+ .withDescription("Interval at which to try committing offsets, given in milliseconds. Defaults to 1 minute (60,000 ms).")
.withDefault(60000L)
.withValidation(Field::isNonNegativeInteger);
/**
- * An optional advanced field that specifies the maximum amount of time that the embedded connector should wait
- * for an offset commit to complete.
+ * An optional advanced field that specifies the maximum amount of time that the embedded connector should wait for
+ * an offset commit to complete.
*/
public static final Field OFFSET_COMMIT_TIMEOUT_MS = Field.create("offset.flush.timeout.ms")
- .withDescription("Maximum number of milliseconds to wait for records to flush and partition offset data to be"
+ .withDescription("Time to wait for records to flush and partition offset data to be"
+ " committed to offset storage before cancelling the process and restoring the offset "
- + "data to be committed in a future attempt.")
+ + "data to be committed in a future attempt, given in milliseconds. Defaults to 5 seconds (5000 ms).")
.withDefault(5000L)
.withValidation(Field::isPositiveInteger);
@@ -209,8 +209,7 @@ public final class EmbeddedEngine implements DebeziumEngine {
INTERNAL_KEY_CONVERTER_CLASS, INTERNAL_VALUE_CONVERTER_CLASS);
/**
- * How long we wait before forcefully stopping the connector thread when
- * shutting down. Must be longer than
+ * How long we wait before forcefully stopping the connector thread when shutting down. Must be longer than
* {@link ChangeEventSourceCoordinator#SHUTDOWN_WAIT_TIMEOUT} * 2.
*/
private static final Duration WAIT_FOR_COMPLETION_BEFORE_INTERRUPT_DEFAULT = Duration.ofMinutes(5);
@@ -368,8 +367,8 @@ public final class EmbeddedEngine implements DebeziumEngine {
}
/**
- * Causes the current thread to wait until the {@link #handle(boolean, String, Throwable) completion occurs}
- * or until the thread is {@linkplain Thread#interrupt interrupted}.
+ * Causes the current thread to wait until the {@link #handle(boolean, String, Throwable) completion occurs} or
+ * until the thread is {@linkplain Thread#interrupt interrupted}.
*
* This method returns immediately if the connector has completed already.
*
@@ -387,8 +386,8 @@ public final class EmbeddedEngine implements DebeziumEngine {
*
* @param timeout the maximum time to wait
* @param unit the time unit of the {@code timeout} argument
- * @return {@code true} if the completion was received, or {@code false} if the waiting time elapsed before the completion
- * was received.
+ * @return {@code true} if the completion was received, or {@code false} if the waiting time elapsed before the
+ * completion was received.
* @throws InterruptedException if the current thread is interrupted while waiting
*/
public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
@@ -398,8 +397,8 @@ public final class EmbeddedEngine implements DebeziumEngine {
/**
* Determine if the connector has completed.
*
- * @return {@code true} if the connector has completed, or {@code false} if the connector is still running and this
- * callback has not yet been {@link #handle(boolean, String, Throwable) notified}
+ * @return {@code true} if the connector has completed, or {@code false} if the connector is still running and
+ * this callback has not yet been {@link #handle(boolean, String, Throwable) notified}
*/
public boolean hasCompleted() {
return completed.getCount() == 0;
@@ -408,8 +407,9 @@ public final class EmbeddedEngine implements DebeziumEngine {
/**
* Get whether the connector completed normally.
*
- * @return {@code true} if the connector completed normally, or {@code false} if the connector produced an error that
- * prevented startup or premature termination (or the connector has not yet {@link #hasCompleted() completed})
+ * @return {@code true} if the connector completed normally, or {@code false} if the connector produced an error
+ * that prevented startup or premature termination (or the connector has not yet {@link #hasCompleted()
+ * completed})
*/
public boolean success() {
return success;
@@ -427,7 +427,8 @@ public final class EmbeddedEngine implements DebeziumEngine {
/**
* Get the completion error, if there is one.
*
- * @return the completion error, or null if there is no error or connector has not yet {@link #hasCompleted() completed}
+ * @return the completion error, or null if there is no error or connector has not yet {@link #hasCompleted()
+ * completed}
*/
public Throwable error() {
return error;
@@ -437,7 +438,7 @@ public final class EmbeddedEngine implements DebeziumEngine {
* Determine if there is a completion error.
*
* @return {@code true} if there is a {@link #error completion error}, or {@code false} if there is no error or
- * the connector has not yet {@link #hasCompleted() completed}
+ * the connector has not yet {@link #hasCompleted() completed}
*/
public boolean hasError() {
return error != null;
@@ -461,40 +462,15 @@ public final class EmbeddedEngine implements DebeziumEngine {
public static interface ChangeConsumer extends DebeziumEngine.ChangeConsumer {
}
- protected class SourceRecordOffsets implements DebeziumEngine.Offsets {
-
- private final HashMap offsets = new HashMap<>();
-
- /**
- * Performs {@link HashMap#put(Object, Object)} on the offsets map.
- *
- * @param key key with which to put the value
- * @param value value to be put with the key
- */
- @Override
- public void set(String key, Object value) {
- offsets.put(key, value);
- }
-
- /**
- * Retrieves the offsets map.
- *
- * @return HashMap of the offsets
- */
- protected HashMap getOffsets() {
- return offsets;
- }
- }
-
private static ChangeConsumer buildDefaultChangeConsumer(Consumer consumer) {
return new ChangeConsumer() {
/**
* the default implementation that is compatible with the old Consumer api.
*
- * On every record, it calls the consumer, and then only marks the record
- * as processed when accept returns, additionally, it handles StopConnectorExceptions
- * and ensures that we all ways try and mark a batch as finished, even with exceptions
+ * On every record, it calls the consumer, and then only marks the record as processed when accept returns,
+ * additionally, it handles StopConnectorExceptions and ensures that we all ways try and mark a batch as
+ * finished, even with exceptions
* @param records the records to be processed
* @param committer the committer that indicates to the system that we are finished
*
@@ -534,8 +510,8 @@ public final class EmbeddedEngine implements DebeziumEngine {
Builder using(Configuration config);
/**
- * Use the specified clock when needing to determine the current time. Passing null or not calling this
- * method results in the connector using the {@link Clock#system() system clock}.
+ * Use the specified clock when needing to determine the current time. Passing null or not calling
+ * this method results in the connector using the {@link Clock#system() system clock}.
*
* @param clock the clock
* @return this builder object so methods can be chained together; never null
@@ -667,19 +643,19 @@ public final class EmbeddedEngine implements DebeziumEngine {
}
/**
- * Run this embedded connector and deliver database changes to the registered {@link Consumer}. This method blocks until
- * the connector is stopped.
+ * Run this embedded connector and deliver database changes to the registered {@link Consumer}. This method blocks
+ * until the connector is stopped.
*
- * First, the method checks to see if this instance is currently {@link #run() running}, and if so immediately returns.
+ * First, the method checks to see if this instance is currently {@link #run() 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 {@link 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
- * {@link #stop() stopped}.
+ * 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 {@link 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 {@link #stop() stopped}.
*
- * Note that there are two ways to stop a connector running on a thread: calling {@link #stop()} from another thread, or
- * interrupting the thread (e.g., via {@link ExecutorService#shutdownNow()}).
+ * Note that there are two ways to stop a connector running on a thread: calling {@link #stop()} from another
+ * thread, or interrupting the thread (e.g., via {@link ExecutorService#shutdownNow()}).
*
* This method can be called repeatedly as needed.
*/
@@ -710,18 +686,19 @@ public final class EmbeddedEngine implements DebeziumEngine {
return;
}
- //// Instantiate the offset store ...
- //final String offsetStoreClassName = config.getString(OFFSET_STORAGE);
- //OffsetBackingStore offsetStore = null;
- //try {
- // @SuppressWarnings("unchecked")
- // Class offsetStoreClass = (Class) classLoader.loadClass(offsetStoreClassName);
- // offsetStore = offsetStoreClass.getDeclaredConstructor().newInstance();
- //}
- //catch (Throwable t) {
- // fail("Unable to instantiate OffsetBackingStore class '" + offsetStoreClassName + "'", t);
- // return;
- //}
+ // // Instantiate the offset store ...
+ // final String offsetStoreClassName = config.getString(OFFSET_STORAGE);
+ // OffsetBackingStore offsetStore = null;
+ // try {
+ // @SuppressWarnings("unchecked")
+ // Class offsetStoreClass = (Class) classLoader
+ // .loadClass(offsetStoreClassName);
+ // offsetStore = offsetStoreClass.getDeclaredConstructor().newInstance();
+ // }
+ // catch (Throwable t) {
+ // fail("Unable to instantiate OffsetBackingStore class '" + offsetStoreClassName + "'", t);
+ // return;
+ // }
// Initialize the offset store ...
try {
@@ -762,7 +739,7 @@ public final class EmbeddedEngine implements DebeziumEngine {
try {
// Start the connector with the given properties and get the task configurations ...
- connector.start(config.asMap());
+ connector.start(workerConfig.originalsStrings());
connectorCallback.ifPresent(DebeziumEngine.ConnectorCallback::connectorStarted);
List> taskConfigs = connector.taskConfigs(1);
Class taskClass = connector.taskClass();
@@ -798,6 +775,14 @@ public final class EmbeddedEngine implements DebeziumEngine {
connectorCallback.ifPresent(DebeziumEngine.ConnectorCallback::taskStarted);
}
catch (Throwable t) {
+ // Clean-up allocated resources
+ try {
+ LOGGER.debug("Stopping the task");
+ task.stop();
+ }
+ catch (Throwable tstop) {
+ LOGGER.info("Error while trying to stop the task");
+ }
// Mask the passwords ...
Configuration config = Configuration.from(taskConfigs.get(0)).withMaskedPasswords();
String msg = "Unable to initialize and start connector's task class '" + taskClass.getName() + "' with config: "
@@ -873,7 +858,7 @@ public final class EmbeddedEngine implements DebeziumEngine {
}
try {
// First stop the task ...
- LOGGER.debug("Stopping the task and engine");
+ LOGGER.info("Stopping the task and engine");
task.stop();
connectorCallback.ifPresent(DebeziumEngine.ConnectorCallback::taskStopped);
// Always commit offsets that were captured from the source records we actually processed ...
@@ -883,6 +868,10 @@ public final class EmbeddedEngine implements DebeziumEngine {
succeed("Connector '" + connectorClassName + "' completed normally.");
}
}
+ catch (InterruptedException e) {
+ LOGGER.debug("Interrupted while committing offsets");
+ Thread.currentThread().interrupt();
+ }
catch (Throwable t) {
fail("Error while trying to stop the task and commit the offsets", t);
}
@@ -920,8 +909,7 @@ public final class EmbeddedEngine implements DebeziumEngine {
}
/**
- * Creates a new RecordCommitter that is responsible for informing the engine
- * about the updates to the given batch
+ * Creates a new RecordCommitter that is responsible for informing the engine about the updates to the given batch
* @param offsetWriter the offsetWriter current in use
* @param task the sourcetask
* @param commitTimeout the time in ms until a commit times out
@@ -958,6 +946,34 @@ public final class EmbeddedEngine implements DebeziumEngine {
};
}
+ /**
+ * Implementation of {@link DebeziumEngine.Offsets} which can be used to construct a {@link SourceRecord} with its
+ * offsets.
+ */
+ protected class SourceRecordOffsets implements DebeziumEngine.Offsets {
+ private HashMap offsets = new HashMap<>();
+
+ /**
+ * Performs {@link HashMap#put(Object, Object)} on the offsets map.
+ *
+ * @param key key with which to put the value
+ * @param value value to be put with the key
+ */
+ @Override
+ public void set(String key, Object value) {
+ offsets.put(key, value);
+ }
+
+ /**
+ * Retrieves the offsets map.
+ *
+ * @return HashMap of the offsets
+ */
+ protected HashMap getOffsets() {
+ return offsets;
+ }
+ }
+
/**
* Determine if we should flush offsets to storage, and if so then attempt to flush offsets.
*
@@ -967,7 +983,8 @@ public final class EmbeddedEngine implements DebeziumEngine {
* @param task the task which produced the records for which the offsets have been committed
*/
protected void maybeFlush(OffsetStorageWriter offsetWriter, OffsetCommitPolicy policy, Duration commitTimeout,
- SourceTask task) {
+ SourceTask task)
+ throws InterruptedException {
// Determine if we need to commit to offset storage ...
long timeSinceLastCommitMillis = clock.currentTimeInMillis() - timeOfLastCommitMillis;
if (policy.performCommit(recordsSinceLastCommit, Duration.ofMillis(timeSinceLastCommitMillis))) {
@@ -982,7 +999,8 @@ public final class EmbeddedEngine implements DebeziumEngine {
* @param commitTimeout the timeout to wait for commit results
* @param task the task which produced the records for which the offsets have been committed
*/
- protected void commitOffsets(OffsetStorageWriter offsetWriter, Duration commitTimeout, SourceTask task) {
+ protected void commitOffsets(OffsetStorageWriter offsetWriter, Duration commitTimeout, SourceTask task)
+ throws InterruptedException {
long started = clock.currentTimeInMillis();
long timeout = started + commitTimeout.toMillis();
if (!offsetWriter.beginFlush()) {
@@ -1004,6 +1022,14 @@ public final class EmbeddedEngine implements DebeziumEngine {
catch (InterruptedException e) {
LOGGER.warn("Flush of {} offsets interrupted, cancelling", this);
offsetWriter.cancelFlush();
+
+ if (this.runningThread.get() == Thread.currentThread()) {
+ // this thread is still set as the running thread -> we were not interrupted
+ // due the stop() call -> probably someone else called the interrupt on us ->
+ // -> we should raise the interrupt flag
+ Thread.currentThread().interrupt();
+ throw e;
+ }
}
catch (ExecutionException e) {
LOGGER.error("Flush of {} offsets threw an unexpected exception: ", this, e);
@@ -1028,8 +1054,8 @@ public final class EmbeddedEngine implements DebeziumEngine {
* Stop the execution of this embedded connector. This method does not block until the connector is stopped; use
* {@link #await(long, TimeUnit)} for this purpose.
*
- * @return {@code true} if the connector was {@link #run() running} and will eventually stop, or {@code false} if it was not
- * running when this method is called
+ * @return {@code true} if the connector was {@link #run() running} and will eventually stop, or {@code false} if it
+ * was not running when this method is called
* @see #await(long, TimeUnit)
*/
public boolean stop() {
@@ -1060,14 +1086,14 @@ public final class EmbeddedEngine implements DebeziumEngine {
}
/**
- * Wait for the connector to complete processing. If the processor is not running, this method returns immediately; however,
- * if the processor is {@link #stop() stopped} and restarted before this method is called, this method will return only
- * when it completes the second time.
+ * Wait for the connector to complete processing. If the processor is not running, this method returns immediately;
+ * however, if the processor is {@link #stop() stopped} and restarted before this method is called, this method will
+ * return only when it completes the second time.
*
* @param timeout the maximum amount of time to wait before returning
* @param unit the unit of time; may not be null
- * @return {@code true} if the connector completed within the timeout (or was not running), or {@code false} if it is still
- * running when the timeout occurred
+ * @return {@code true} if the connector completed within the timeout (or was not running), or {@code 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
*/
public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
diff --git a/supplier/cdc-debezium-supplier/pom.xml b/supplier/cdc-debezium-supplier/pom.xml
index 09b4c0e4..439be64c 100644
--- a/supplier/cdc-debezium-supplier/pom.xml
+++ b/supplier/cdc-debezium-supplier/pom.xml
@@ -14,7 +14,7 @@
CDC Debezium Suppliers
- 1.7.1.Final
+ 1.9.6.Final