diff --git a/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcMySqlTestSupport.java b/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcMySqlTestSupport.java index 00c68f15..7e0d990e 100644 --- a/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcMySqlTestSupport.java +++ b/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcMySqlTestSupport.java @@ -33,7 +33,7 @@ public abstract class CdcMySqlTestSupport { static String MAPPED_PORT; - static GenericContainer debeziumMySQL = new GenericContainer<>("debezium/example-mysql:1.7.1.Final") + static GenericContainer debeziumMySQL = new GenericContainer<>("debezium/example-mysql:1.9.6.Final") .withEnv("MYSQL_ROOT_PASSWORD", "debezium") .withEnv("MYSQL_USER", "mysqluser") .withEnv("MYSQL_PASSWORD", "mysqlpw") diff --git a/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcSourceDatabasesIntegrationTest.java b/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcSourceDatabasesIntegrationTest.java index add32b25..675b4a90 100644 --- a/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcSourceDatabasesIntegrationTest.java +++ b/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcSourceDatabasesIntegrationTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2021 the original author or authors. + * Copyright 2020-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -20,6 +20,8 @@ import java.time.Duration; import java.util.ArrayList; import java.util.List; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.awaitility.Awaitility; import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Tag; @@ -38,7 +40,7 @@ import org.springframework.messaging.Message; import org.springframework.util.CollectionUtils; import static org.assertj.core.api.Assertions.assertThat; -import static org.springframework.cloud.stream.app.source.cdc.CdcTestUtils.receiveAll; + /** * @author Christian Tzolov @@ -48,6 +50,15 @@ import static org.springframework.cloud.stream.app.source.cdc.CdcTestUtils.recei @Tag("integration") public class CdcSourceDatabasesIntegrationTest { + + private static final String DEBEZIUM_EXAMPLE_MONGODB_1_9_6_FINAL = "debezium/example-mongodb:1.9.6.Final"; + + private static final String DEBEZIUM_EXAMPLE_POSTGRES_1_9_6_FINAL = "debezium/example-postgres:1.9.6.Final"; + + private static final String DEBEZIUM_EXAMPLE_MYSQL_1_9_6_FINAL = "debezium/example-mysql:1.9.6.Final"; + + private static final Log logger = LogFactory.getLog(CdcSourceDatabasesIntegrationTest.class); + private final SpringApplicationBuilder applicationBuilder = new SpringApplicationBuilder( TestChannelBinderConfiguration.getCompleteConfiguration(TestCdcSourceApplication.class)) .web(WebApplicationType.NONE) @@ -57,13 +68,12 @@ public class CdcSourceDatabasesIntegrationTest { "cdc.schema=false", "cdc.flattening.enabled=true", "cdc.stream.header.offset=true", - // "cdc.config.database.server.id=85744", "cdc.config.database.server.name=my-app-connector", "cdc.config.database.history=io.debezium.relational.history.MemoryDatabaseHistory"); @Test public void mysql() { - GenericContainer debeziumMySQL = new GenericContainer<>("debezium/example-mysql:1.7.1.Final") + GenericContainer debeziumMySQL = new GenericContainer<>(DEBEZIUM_EXAMPLE_MYSQL_1_9_6_FINAL) .withEnv("MYSQL_ROOT_PASSWORD", "debezium") .withEnv("MYSQL_USER", "mysqluser") .withEnv("MYSQL_PASSWORD", "mysqlpw") @@ -81,8 +91,10 @@ public class CdcSourceDatabasesIntegrationTest { "--cdc.config.database.port=" + MAPPED_PORT)) { OutputDestination outputDestination = context.getBean(OutputDestination.class); // Using local region here - List> messages = receiveAll(outputDestination); + List> messages = CdcTestUtils.receiveAll(outputDestination); assertThat(messages).isNotNull(); + // Message size should correspond to the number of insert statements in the sample inventor DB configured in the debezium/example-mysql:1.9.6.Final: + // https://github.com/debezium/container-images/blob/main/examples/mysql/1.9/inventory.sql assertThat(messages).hasSizeGreaterThanOrEqualTo(52); } } @@ -116,7 +128,7 @@ public class CdcSourceDatabasesIntegrationTest { "--cdc.config.database.port=" + sqlServer.getMappedPort(1433))) { OutputDestination outputDestination = context.getBean(OutputDestination.class); // Using local region here - List> messages = receiveAll(outputDestination); + List> messages = CdcTestUtils.receiveAll(outputDestination); assertThat(messages).isNotNull(); assertThat(messages).hasSize(30); } @@ -124,7 +136,7 @@ public class CdcSourceDatabasesIntegrationTest { @Test public void postgres() { - GenericContainer postgres = new GenericContainer("debezium/example-postgres:1.7.1.Final") + GenericContainer postgres = new GenericContainer(DEBEZIUM_EXAMPLE_POSTGRES_1_9_6_FINAL) .withEnv("POSTGRES_USER", "postgres") .withEnv("POSTGRES_PASSWORD", "postgres") .withExposedPorts(5432); @@ -137,18 +149,21 @@ public class CdcSourceDatabasesIntegrationTest { "--cdc.config.slot.name=debezium", "--cdc.config.database.dbname=postgres", "--cdc.config.database.hostname=localhost", + // "--cdc.config.table.include.list=inventory.*", "--cdc.config.database.port=" + postgres.getMappedPort(5432))) { OutputDestination outputDestination = context.getBean(OutputDestination.class); // Using local region here List> allMessages = new ArrayList<>(); Awaitility.await().atMost(Duration.ofMinutes(5)).until(() -> { - List> messageChunk = receiveAll(outputDestination); + List> messageChunk = CdcTestUtils.receiveAll(outputDestination); if (!CollectionUtils.isEmpty(messageChunk)) { - System.out.println("Chunk size: " + messageChunk.size()); + logger.info("Chunk size: " + messageChunk.size()); allMessages.addAll(messageChunk); } - return allMessages.size() == 5786; + // Message size should correspond to the number of insert statements in the sample inventor DB configured in the debezium/example-postgres:1.9.6.Final: + // https://github.com/debezium/container-images/blob/main/examples/postgres/1.9/inventory.sql + return allMessages.size() == 29; // Inventory DB entries }); } postgres.stop(); @@ -157,7 +172,7 @@ public class CdcSourceDatabasesIntegrationTest { @Test @Disabled public void mongodb() { - GenericContainer mongodb = new GenericContainer("debezium/example-mongodb:1.7.1.Final") + GenericContainer mongodb = new GenericContainer(DEBEZIUM_EXAMPLE_MONGODB_1_9_6_FINAL) .withEnv("MONGODB_USER", "debezium") .withEnv("MONGODB_PASSWORD", "dbz") .withExposedPorts(27017); @@ -172,7 +187,7 @@ public class CdcSourceDatabasesIntegrationTest { "--cdc.config.collection.include.list=inventory[.]*")) { OutputDestination outputDestination = context.getBean(OutputDestination.class); // Using local region here - List> messages = receiveAll(outputDestination); + List> messages = CdcTestUtils.receiveAll(outputDestination); assertThat(messages).isNotNull(); assertThat(messages).hasSize(666); } diff --git a/functions/common/cdc-debezium-boot-starter/pom.xml b/functions/common/cdc-debezium-boot-starter/pom.xml index 02fb1477..fce6e9b0 100644 --- a/functions/common/cdc-debezium-boot-starter/pom.xml +++ b/functions/common/cdc-debezium-boot-starter/pom.xml @@ -15,7 +15,7 @@ Change Data Capture (CDC) Debezium Boot Starter - 1.7.1.Final + 1.9.6.Final diff --git a/functions/common/cdc-debezium-boot-starter/src/test/java/org/springframework/cloud/fn/common/cdc/CdcBootStarterIntegrationTest.java b/functions/common/cdc-debezium-boot-starter/src/test/java/org/springframework/cloud/fn/common/cdc/CdcBootStarterIntegrationTest.java index 58ef862a..64c041ed 100644 --- a/functions/common/cdc-debezium-boot-starter/src/test/java/org/springframework/cloud/fn/common/cdc/CdcBootStarterIntegrationTest.java +++ b/functions/common/cdc-debezium-boot-starter/src/test/java/org/springframework/cloud/fn/common/cdc/CdcBootStarterIntegrationTest.java @@ -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") diff --git a/functions/common/cdc-debezium-common/pom.xml b/functions/common/cdc-debezium-common/pom.xml index bc210061..fc447e8f 100644 --- a/functions/common/cdc-debezium-common/pom.xml +++ b/functions/common/cdc-debezium-common/pom.xml @@ -1,6 +1,6 @@ - + 4.0.0 @@ -15,10 +15,15 @@ Change Data Capture (CDC) Debezium Common - 1.7.1.Final + 1.9.6.Final + + io.debezium + debezium-api + ${version.debezium} + io.debezium debezium-embedded diff --git a/functions/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/EmbeddedEngine.java b/functions/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/EmbeddedEngine.java index 4f281416..3b896f66 100644 --- a/functions/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/EmbeddedEngine.java +++ b/functions/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/functions/supplier/cdc-debezium-supplier/pom.xml b/functions/supplier/cdc-debezium-supplier/pom.xml index 09b4c0e4..439be64c 100644 --- a/functions/supplier/cdc-debezium-supplier/pom.xml +++ b/functions/supplier/cdc-debezium-supplier/pom.xml @@ -14,7 +14,7 @@ CDC Debezium Suppliers - 1.7.1.Final + 1.9.6.Final