SpringCloud/spring-functions-catalog
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Update CDC Debezium to 1.9.6

Browse Source 
- The update fixes integration tests
This commit is contained in:
Chris Bono
2022-10-25 23:09:59 -05:00
parent afd5f82ec4
commit 378ac48e6a
5 changed files with 133 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
common/cdc-debezium-boot-starter/pom.xml
View File

@@ -15,7 +15,7 @@
<description>Change Data Capture (CDC) Debezium Boot Starter</description>
<properties>
<version.debezium>1.7.1.Final</version.debezium>
<version.debezium>1.9.6.Final</version.debezium>
</properties>
<dependencies>

2
common/cdc-debezium-boot-starter/src/test/java/org/springframework/cloud/fn/common/cdc/CdcBootStarterIntegrationTest.java
View File

@@ -50,7 +50,7 @@ public class CdcBootStarterIntegrationTest {
@Container
static GenericContainer debeziumMySQL =
new GenericContainer<>(DockerImageName.parse("debezium/example-mysql:1.7.1.Final"))
new GenericContainer<>(DockerImageName.parse("debezium/example-mysql:1.9.6.Final"))
.withEnv("MYSQL_ROOT_PASSWORD", "debezium")
.withEnv("MYSQL_USER", "mysqluser")
.withEnv("MYSQL_PASSWORD", "mysqlpw")

11
common/cdc-debezium-common/pom.xml
View File

@@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
@@ -15,10 +15,15 @@
<description>Change Data Capture (CDC) Debezium Common</description>
<properties>
<version.debezium>1.7.1.Final</version.debezium>
<version.debezium>1.9.6.Final</version.debezium>
</properties>
<dependencies>
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-api</artifactId>
<version>${version.debezium}</version>
</dependency>
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-embedded</artifactId>

218
common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/EmbeddedEngine.java
View File

@@ -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.
* <p>
* 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.
* <p>
* 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<SourceRecord> {
*/
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<SourceRecord> {
+ 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<SourceRecord> {
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<SourceRecord> {
}
/**
* 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}.
* <p>
* This method returns immediately if the connector has completed already.
*
@@ -387,8 +386,8 @@ public final class EmbeddedEngine implements DebeziumEngine<SourceRecord> {
*
* @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<SourceRecord> {
/**
* 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<SourceRecord> {
/**
* 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<SourceRecord> {
/**
* 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<SourceRecord> {
* 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<SourceRecord> {
public static interface ChangeConsumer extends DebeziumEngine.ChangeConsumer<SourceRecord> {
}
protected class SourceRecordOffsets implements DebeziumEngine.Offsets {
private final HashMap<String, Object> 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<String, Object> getOffsets() {
return offsets;
}
}
private static ChangeConsumer buildDefaultChangeConsumer(Consumer<SourceRecord> 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<SourceRecord> {
Builder using(Configuration config);
/**
* Use the specified clock when needing to determine the current time. Passing <code>null</code> 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 <code>null</code> 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<SourceRecord> {
}
/**
* 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.
* <p>
* 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.
* <p>
* 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}.
* <p>
* 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()}).
* <p>
* This method can be called repeatedly as needed.
*/
@@ -710,18 +686,19 @@ public final class EmbeddedEngine implements DebeziumEngine<SourceRecord> {
return;
}
//// Instantiate the offset store ...
//final String offsetStoreClassName = config.getString(OFFSET_STORAGE);
//OffsetBackingStore offsetStore = null;
//try {
// @SuppressWarnings("unchecked")
// Class<? extends OffsetBackingStore> offsetStoreClass = (Class<OffsetBackingStore>) 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<? extends OffsetBackingStore> offsetStoreClass = (Class<OffsetBackingStore>) 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<SourceRecord> {
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<Map<String, String>> taskConfigs = connector.taskConfigs(1);
Class<? extends Task> taskClass = connector.taskClass();
@@ -798,6 +775,14 @@ public final class EmbeddedEngine implements DebeziumEngine<SourceRecord> {
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<SourceRecord> {
}
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<SourceRecord> {
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<SourceRecord> {
}
/**
* 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<SourceRecord> {
};
}
/**
* 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<String, Object> 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<String, Object> 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<SourceRecord> {
* @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<SourceRecord> {
* @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<SourceRecord> {
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<SourceRecord> {
* 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<SourceRecord> {
}
/**
* 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 {

2
supplier/cdc-debezium-supplier/pom.xml
View File

@@ -14,7 +14,7 @@
<description>CDC Debezium Suppliers</description>
<properties>
<version.debezium>1.7.1.Final</version.debezium>
<version.debezium>1.9.6.Final</version.debezium>
</properties>
<dependencies>