From 190fa374d6f0cae420364e3d2872a69003f79a0f Mon Sep 17 00:00:00 2001 From: Christian Tzolov Date: Tue, 2 Aug 2022 18:24:42 +0200 Subject: [PATCH] Upgrade CDC from 1.7.1 to 1.9.5 - update the cdc-common, cdc-supplier and cdc-source to the latest 1.9.5.Final version. - adjust IT tests - fix minor checkstyle issues (accross all apps) Resolves #188 Signed-off-by: Christian Tzolov --- .gitignore | 1 + .../cdc/CdcFlatteningIntegrationTest.java | 65 +++-- .../app/source/cdc/CdcMySqlTestSupport.java | 2 +- .../CdcSourceDatabasesIntegrationTest.java | 18 +- .../common/cdc-debezium-boot-starter/pom.xml | 2 +- .../cdc/CdcBootStarterIntegrationTest.java | 2 +- functions/common/cdc-debezium-common/pom.xml | 2 +- .../fn/common/cdc/CdcCommonConfiguration.java | 4 +- .../cloud/fn/common/cdc/EmbeddedEngine.java | 223 ++++++++++-------- .../HttpRequestFunctionConfiguration.java | 1 - .../supplier/cdc-debezium-supplier/pom.xml | 2 +- .../mail/MailSupplierConfiguration.java | 1 - 12 files changed, 174 insertions(+), 149 deletions(-) diff --git a/.gitignore b/.gitignore index 6774efa8..54bf7798 100644 --- a/.gitignore +++ b/.gitignore @@ -28,3 +28,4 @@ coverage-error.log .apt_generated aws.credentials.properties .flattened-pom.xml +.vscode diff --git a/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcFlatteningIntegrationTest.java b/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcFlatteningIntegrationTest.java index 7281bdc4..2e60cd75 100644 --- a/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcFlatteningIntegrationTest.java +++ b/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcFlatteningIntegrationTest.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. @@ -18,6 +18,7 @@ package org.springframework.cloud.stream.app.source.cdc; import java.util.List; +import net.javacrumbs.jsonunit.JsonAssert; import net.javacrumbs.jsonunit.core.Configuration; import org.junit.jupiter.api.Test; @@ -25,6 +26,7 @@ import org.springframework.boot.test.context.FilteredClassLoader; import org.springframework.boot.test.context.runner.ApplicationContextRunner; import org.springframework.boot.test.context.runner.ContextConsumer; import org.springframework.cloud.fn.common.cdc.CdcCommonProperties; +import org.springframework.cloud.fn.supplier.cdc.CdcSupplierConfiguration; import org.springframework.cloud.stream.binder.test.OutputDestination; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; import org.springframework.context.ApplicationContext; @@ -34,11 +36,8 @@ import org.springframework.test.jdbc.JdbcTestUtils; import org.springframework.util.ClassUtils; import org.springframework.util.StringUtils; -import static net.javacrumbs.jsonunit.JsonAssert.assertJsonEquals; import static org.assertj.core.api.Assertions.assertThat; -import static org.springframework.cloud.fn.supplier.cdc.CdcSupplierConfiguration.ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL; -import static org.springframework.cloud.stream.app.source.cdc.CdcTestUtils.receiveAll; -import static org.springframework.cloud.stream.app.source.cdc.CdcTestUtils.resourceToString; + /** * @author Christian Tzolov @@ -79,24 +78,24 @@ public class CdcFlatteningIntegrationTest extends CdcMySqlTestSupport { final ContextConsumer noFlatteningTest = context -> { OutputDestination outputDestination = context.getBean(OutputDestination.class); - boolean isKafkaPresent = ClassUtils.isPresent(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL, + boolean isKafkaPresent = ClassUtils.isPresent(CdcSupplierConfiguration.ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL, context.getClassLoader()); - List> messages = receiveAll(outputDestination); + List> messages = CdcTestUtils.receiveAll(outputDestination); assertThat(messages).hasSizeGreaterThanOrEqualTo(52); - assertJsonEquals(resourceToString( + JsonAssert.assertJsonEquals(CdcTestUtils.resourceToString( "classpath:/json/mysql_ddl_drop_inventory_address_table.json"), toString(messages.get(1).getPayload()), Configuration.empty().whenIgnoringPaths("schemaName", "tableChanges", "source.sequence", "source.ts_ms")); assertThat(messages.get(1).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector"); - assertJsonEquals("{\"databaseName\":\"inventory\"}", toString(messages.get(1).getHeaders().get("cdc_key"))); + JsonAssert.assertJsonEquals("{\"databaseName\":\"inventory\"}", toString(messages.get(1).getHeaders().get("cdc_key"))); - assertJsonEquals(resourceToString("classpath:/json/mysql_insert_inventory_products_106.json"), + JsonAssert.assertJsonEquals(CdcTestUtils.resourceToString("classpath:/json/mysql_insert_inventory_products_106.json"), toString(messages.get(39).getPayload()), Configuration.empty().whenIgnoringPaths("source.sequence", "source.ts_ms")); assertThat(messages.get(39).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector.inventory.products"); - assertJsonEquals("{\"id\":106}", toString(messages.get(39).getHeaders().get("cdc_key"))); + JsonAssert.assertJsonEquals("{\"id\":106}", toString(messages.get(39).getHeaders().get("cdc_key"))); jdbcTemplate.update( "insert into `customers`(`first_name`,`last_name`,`email`) VALUES('Test666', 'Test666', 'Test666@spring.org')"); @@ -105,28 +104,28 @@ public class CdcFlatteningIntegrationTest extends CdcMySqlTestSupport { jdbcTemplate.update("UPDATE `customers` SET `last_name`='Test999' WHERE first_name = 'Test666'"); JdbcTestUtils.deleteFromTableWhere(jdbcTemplate, "customers", "first_name = ?", "Test666"); - messages = receiveAll(outputDestination); + messages = CdcTestUtils.receiveAll(outputDestination); assertThat(messages).hasSize(isKafkaPresent ? 4 : 3); - assertJsonEquals(resourceToString("classpath:/json/mysql_update_inventory_customers.json"), + JsonAssert.assertJsonEquals(CdcTestUtils.resourceToString("classpath:/json/mysql_update_inventory_customers.json"), toString(messages.get(1).getPayload()), Configuration.empty().whenIgnoringPaths("source.sequence")); assertThat(messages.get(1).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector.inventory.customers"); - assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(1).getHeaders().get("cdc_key"))); + JsonAssert.assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(1).getHeaders().get("cdc_key"))); - assertJsonEquals(resourceToString("classpath:/json/mysql_delete_inventory_customers.json"), + JsonAssert.assertJsonEquals(CdcTestUtils.resourceToString("classpath:/json/mysql_delete_inventory_customers.json"), toString(messages.get(2).getPayload()), Configuration.empty().whenIgnoringPaths("source.sequence")); assertThat(messages.get(1).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector.inventory.customers"); - assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(1).getHeaders().get("cdc_key"))); + JsonAssert.assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(1).getHeaders().get("cdc_key"))); if (isKafkaPresent) { assertThat(messages.get(3).getPayload().getClass().getCanonicalName()) - .isEqualTo(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL, + .isEqualTo(CdcSupplierConfiguration.ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL, "Tombstones event should have KafkaNull payload"); assertThat(messages.get(3).getHeaders().get("cdc_topic")) .isEqualTo("my-app-connector.inventory.customers"); - assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(3).getHeaders().get("cdc_key"))); + JsonAssert.assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(3).getHeaders().get("cdc_key"))); } }; @@ -166,32 +165,32 @@ public class CdcFlatteningIntegrationTest extends CdcMySqlTestSupport { final ContextConsumer flatteningTest = context -> { OutputDestination outputDestination = context.getBean(OutputDestination.class); - boolean isKafkaPresent = ClassUtils.isPresent(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL, + boolean isKafkaPresent = ClassUtils.isPresent(CdcSupplierConfiguration.ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL, context.getClassLoader()); - List> messages = receiveAll(outputDestination); + List> messages = CdcTestUtils.receiveAll(outputDestination); assertThat(messages).hasSizeGreaterThanOrEqualTo(52); CdcCommonProperties.Flattening flatteningProps = context.getBean(CdcCommonProperties.class).getFlattening(); - assertJsonEquals(resourceToString( + JsonAssert.assertJsonEquals(CdcTestUtils.resourceToString( "classpath:/json/mysql_ddl_drop_inventory_address_table.json"), toString(messages.get(1).getPayload()), Configuration.empty().whenIgnoringPaths("schemaName", "tableChanges", "source.sequence", "source.ts_ms")); assertThat(messages.get(1).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector"); - assertJsonEquals("{\"databaseName\":\"inventory\"}", + JsonAssert.assertJsonEquals("{\"databaseName\":\"inventory\"}", toString(messages.get(1).getHeaders().get("cdc_key"))); if (flatteningProps.isEnabled()) { - assertJsonEquals(resourceToString("classpath:/json/mysql_flattened_insert_inventory_products_106.json"), + JsonAssert.assertJsonEquals(CdcTestUtils.resourceToString("classpath:/json/mysql_flattened_insert_inventory_products_106.json"), toString(messages.get(39).getPayload())); } else { - assertJsonEquals(resourceToString("classpath:/json/mysql_insert_inventory_products_106.json"), + JsonAssert.assertJsonEquals(CdcTestUtils.resourceToString("classpath:/json/mysql_insert_inventory_products_106.json"), toString(messages.get(39).getPayload())); } assertThat(messages.get(39).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector.inventory.products"); - assertJsonEquals("{\"id\":106}", toString(messages.get(39).getHeaders().get("cdc_key"))); + JsonAssert.assertJsonEquals("{\"id\":106}", toString(messages.get(39).getHeaders().get("cdc_key"))); if (flatteningProps.isEnabled() && flatteningProps.getAddHeaders().contains("op")) { assertThat(messages.get(39).getHeaders().get("__op")).isEqualTo("r"); @@ -204,24 +203,24 @@ public class CdcFlatteningIntegrationTest extends CdcMySqlTestSupport { jdbcTemplate.update("UPDATE `customers` SET `last_name`='Test999' WHERE first_name = 'Test666'"); JdbcTestUtils.deleteFromTableWhere(jdbcTemplate, "customers", "first_name = ?", "Test666"); - messages = receiveAll(outputDestination); + messages = CdcTestUtils.receiveAll(outputDestination); assertThat(messages).hasSize((!flatteningProps.isDropTombstones() && isKafkaPresent) ? 4 : 3); - assertJsonEquals(resourceToString("classpath:/json/mysql_flattened_update_inventory_customers.json"), + JsonAssert.assertJsonEquals(CdcTestUtils.resourceToString("classpath:/json/mysql_flattened_update_inventory_customers.json"), toString(messages.get(1).getPayload())); assertThat(messages.get(1).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector.inventory.customers"); - assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(1).getHeaders().get("cdc_key"))); - if (!StringUtils.isEmpty(flatteningProps.getAddHeaders()) && flatteningProps.getAddHeaders().contains("op")) { + JsonAssert.assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(1).getHeaders().get("cdc_key"))); + if (!StringUtils.hasText(flatteningProps.getAddHeaders()) && flatteningProps.getAddHeaders().contains("op")) { assertThat(messages.get(1).getHeaders().get("__op")).isEqualTo("u"); } if (flatteningProps.getDeleteHandlingMode() == CdcCommonProperties.DeleteHandlingMode.none) { assertThat(toString(messages.get(2).getPayload())).isEqualTo("null"); assertThat(messages.get(1).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector.inventory.customers"); - assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(1).getHeaders().get("cdc_key"))); - if (!StringUtils.isEmpty(flatteningProps.getAddHeaders()) + JsonAssert.assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(1).getHeaders().get("cdc_key"))); + if (!StringUtils.hasText(flatteningProps.getAddHeaders()) && flatteningProps.getAddHeaders().contains("op")) { assertThat(messages.get(2).getHeaders().get("__op")).isEqualTo("d"); } @@ -229,11 +228,11 @@ public class CdcFlatteningIntegrationTest extends CdcMySqlTestSupport { if (!flatteningProps.isDropTombstones() && isKafkaPresent) { assertThat(messages.get(3).getPayload().getClass().getCanonicalName()) - .isEqualTo(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL, + .isEqualTo(CdcSupplierConfiguration.ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL, "Tombstones event should have KafkaNull payload"); assertThat(messages.get(3).getHeaders().get("cdc_topic")) .isEqualTo("my-app-connector.inventory.customers"); - assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(3).getHeaders().get("cdc_key"))); + JsonAssert.assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(3).getHeaders().get("cdc_key"))); } }; 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..1cfdc0a4 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.5.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..dc8e7754 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 @@ -38,7 +38,6 @@ 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 @@ -63,7 +62,7 @@ public class CdcSourceDatabasesIntegrationTest { @Test public void mysql() { - GenericContainer debeziumMySQL = new GenericContainer<>("debezium/example-mysql:1.7.1.Final") + GenericContainer debeziumMySQL = new GenericContainer<>("debezium/example-mysql:1.9.5.Final") .withEnv("MYSQL_ROOT_PASSWORD", "debezium") .withEnv("MYSQL_USER", "mysqluser") .withEnv("MYSQL_PASSWORD", "mysqlpw") @@ -81,7 +80,7 @@ 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(); assertThat(messages).hasSizeGreaterThanOrEqualTo(52); } @@ -116,7 +115,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 +123,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.5.Final") .withEnv("POSTGRES_USER", "postgres") .withEnv("POSTGRES_PASSWORD", "postgres") .withExposedPorts(5432); @@ -137,18 +136,19 @@ 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()); allMessages.addAll(messageChunk); } - return allMessages.size() == 5786; + return allMessages.size() == 29; // Inventory DB entries }); } postgres.stop(); @@ -157,7 +157,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.5.Final") .withEnv("MONGODB_USER", "debezium") .withEnv("MONGODB_PASSWORD", "dbz") .withExposedPorts(27017); @@ -172,7 +172,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 4a546c21..4fb97527 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.5.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..e7019034 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.5.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 dc68cdff..9b6ea394 100644 --- a/functions/common/cdc-debezium-common/pom.xml +++ b/functions/common/cdc-debezium-common/pom.xml @@ -15,7 +15,7 @@ Change Data Capture (CDC) Debezium Common - 1.7.1.Final + 1.9.5.Final diff --git a/functions/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/CdcCommonConfiguration.java b/functions/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/CdcCommonConfiguration.java index 0f4efc4c..e9837cbe 100644 --- a/functions/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/CdcCommonConfiguration.java +++ b/functions/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/CdcCommonConfiguration.java @@ -64,10 +64,10 @@ public class CdcCommonConfiguration { Map config = extractNewRecordState.config().defaultValues(); config.put("drop.tombstones", properties.getFlattening().isDropTombstones()); config.put("delete.handling.mode", properties.getFlattening().getDeleteHandlingMode().name()); - if (!StringUtils.isEmpty(properties.getFlattening().getAddHeaders())) { + if (!StringUtils.hasText(properties.getFlattening().getAddHeaders())) { config.put("add.headers", properties.getFlattening().getAddHeaders()); } - if (!StringUtils.isEmpty(properties.getFlattening().getAddFields())) { + if (!StringUtils.hasText(properties.getFlattening().getAddFields())) { config.put("add.fields", properties.getFlattening().getAddFields()); } 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..0f6663f2 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,29 +151,29 @@ 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); public static final Field OFFSET_COMMIT_POLICY = Field.create("offset.commit.policy") .withDescription("The fully-qualified class name of the commit policy type. This class must implement the interface " - + OffsetCommitPolicy.class.getName() - + ". The default is a periodic commit policy based upon time intervals.") + + OffsetCommitPolicy.class.getName() + + ". The default is a periodic commit policy based upon time intervals.") .withDefault(io.debezium.embedded.spi.OffsetCommitPolicy.PeriodicCommitOffsetPolicy.class.getName()) .withValidation(Field::isClassName); @@ -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 { @@ -1100,3 +1126,4 @@ public final class EmbeddedEngine implements DebeziumEngine { } } } + diff --git a/functions/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionConfiguration.java b/functions/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionConfiguration.java index 7f074e44..f550434e 100644 --- a/functions/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionConfiguration.java +++ b/functions/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionConfiguration.java @@ -23,7 +23,6 @@ import java.util.function.Function; import reactor.core.publisher.Flux; import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.boot.web.reactive.function.client.WebClientCustomizer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.http.HttpHeaders; diff --git a/functions/supplier/cdc-debezium-supplier/pom.xml b/functions/supplier/cdc-debezium-supplier/pom.xml index 737898fd..35b2e656 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.5.Final diff --git a/functions/supplier/mail-supplier/src/main/java/org/springframework/cloud/fn/supplier/mail/MailSupplierConfiguration.java b/functions/supplier/mail-supplier/src/main/java/org/springframework/cloud/fn/supplier/mail/MailSupplierConfiguration.java index 877a44ce..4a71c3b9 100644 --- a/functions/supplier/mail-supplier/src/main/java/org/springframework/cloud/fn/supplier/mail/MailSupplierConfiguration.java +++ b/functions/supplier/mail-supplier/src/main/java/org/springframework/cloud/fn/supplier/mail/MailSupplierConfiguration.java @@ -24,7 +24,6 @@ import java.util.function.Supplier; import javax.mail.URLName; import org.reactivestreams.Publisher; - import reactor.core.publisher.Flux; import org.springframework.beans.factory.annotation.Autowired;