SpringCloud/stream-applications
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-182: Upgrade Debezium to 1.7.1

Browse Source 
Fixes https://github.com/spring-cloud/stream-applications/issues/182

* Upgrade Debezium dependency to 1.7.1
* Remove explicit dependencies for DBs and remove excludes for them from Debezium deps;
rely fully on whatever Debezium connectors bring for us
* Fix `EmbeddedEngine` for compatibility with the Debezium 1.7.1
* Fix CDC tests for the current state of Debezium results
* Fix Checkstyle violations in the MQTT modules
This commit is contained in:
Artem Bilan
2021-11-19 09:28:59 -05:00
committed by Soby Chacko
parent 4c38879288
commit fb5acf51f7
14 changed files with 84 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
applications/sink/mqtt-sink/README.adoc
View File

@@ -24,6 +24,7 @@ $$keep-alive-interval$$:: $$the ping interval in seconds.$$ *($$Integer$$, defau
$$password$$:: $$the password to use when connecting to the broker.$$ *($$String$$, default: `$$guest$$`)*
$$persistence$$:: $$'memory' or 'file'.$$ *($$String$$, default: `$$memory$$`)*
$$persistence-directory$$:: $$Persistence directory.$$ *($$String$$, default: `$$/tmp/paho$$`)*
$$ssl-properties$$:: $$MQTT Client SSL properties.$$ *($$Map<String, String>$$, default: `$$<none>$$`)*
$$url$$:: $$location of the mqtt broker(s) (comma-delimited list).$$ *($$String[]$$, default: `$$[tcp://localhost:1883]$$`)*
$$username$$:: $$the username to use when connecting to the broker.$$ *($$String$$, default: `$$guest$$`)*

40
applications/source/cdc-debezium-source/pom.xml
View File

@@ -26,21 +26,6 @@
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>cdc-debezium-supplier</artifactId>
<version>${java-functions.version}</version>
<exclusions>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
<exclusion>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>${mysql-connector-java.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
@@ -60,10 +45,6 @@
<version>${stream-apps-core.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jdbc</artifactId>
@@ -105,27 +86,6 @@
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.mongodb</groupId>
<artifactId>mongodb-driver</artifactId>
<version>3.12.7</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.mongodb</groupId>
<artifactId>bson</artifactId>
<version>3.12.7</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.mongodb</groupId>
<artifactId>mongodb-driver-core</artifactId>
<version>3.12.7</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>

15
applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcFlatteningIntegrationTest.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-2021 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.core.Configuration;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.FilteredClassLoader;
@@ -42,6 +43,7 @@ import static org.springframework.cloud.stream.app.source.cdc.CdcTestUtils.resou
/**
* @author Christian Tzolov
* @author David Turanski
* @author Artem Bilan
*/
public class CdcFlatteningIntegrationTest extends CdcMySqlTestSupport {
@@ -85,12 +87,14 @@ public class CdcFlatteningIntegrationTest extends CdcMySqlTestSupport {
assertJsonEquals(resourceToString(
"classpath:/json/mysql_ddl_drop_inventory_address_table.json"),
toString(messages.get(1).getPayload()));
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")));
assertJsonEquals(resourceToString("classpath:/json/mysql_insert_inventory_products_106.json"),
toString(messages.get(39).getPayload()));
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")));
@@ -172,7 +176,8 @@ public class CdcFlatteningIntegrationTest extends CdcMySqlTestSupport {
assertJsonEquals(resourceToString(
"classpath:/json/mysql_ddl_drop_inventory_address_table.json"),
toString(messages.get(1).getPayload()));
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")));
@@ -189,7 +194,7 @@ public class CdcFlatteningIntegrationTest extends CdcMySqlTestSupport {
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("c");
assertThat(messages.get(39).getHeaders().get("__op")).isEqualTo("r");
}
jdbcTemplate.update(

4
applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcMySqlTestSupport.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-2021 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.
@@ -33,7 +33,7 @@ public abstract class CdcMySqlTestSupport {
static String MAPPED_PORT;
static GenericContainer debeziumMySQL = new GenericContainer<>("debezium/example-mysql:1.3")
static GenericContainer debeziumMySQL = new GenericContainer<>("debezium/example-mysql:1.7.1.Final")
.withEnv("MYSQL_ROOT_PASSWORD", "debezium")
.withEnv("MYSQL_USER", "mysqluser")
.withEnv("MYSQL_PASSWORD", "mysqlpw")

10
applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcSourceDatabasesIntegrationTest.java
View File

@@ -52,7 +52,7 @@ public class CdcSourceDatabasesIntegrationTest {
TestChannelBinderConfiguration.getCompleteConfiguration(TestCdcSourceApplication.class))
.web(WebApplicationType.NONE)
.properties("spring.cloud.stream.function.definition=cdcSupplier",
"cdc.name=my-sql-connector",
"cdc.name=my-connector",
"cdc.flattening.dropTombstones=false",
"cdc.schema=false",
"cdc.flattening.enabled=true",
@@ -63,7 +63,7 @@ public class CdcSourceDatabasesIntegrationTest {
@Test
public void mysql() {
GenericContainer debeziumMySQL = new GenericContainer<>("debezium/example-mysql:1.3")
GenericContainer debeziumMySQL = new GenericContainer<>("debezium/example-mysql:1.7.1.Final")
.withEnv("MYSQL_ROOT_PASSWORD", "debezium")
.withEnv("MYSQL_USER", "mysqluser")
.withEnv("MYSQL_PASSWORD", "mysqlpw")
@@ -124,7 +124,7 @@ public class CdcSourceDatabasesIntegrationTest {
@Test
public void postgres() {
GenericContainer postgres = new GenericContainer("debezium/example-postgres:1.3")
GenericContainer postgres = new GenericContainer("debezium/example-postgres:1.7.1.Final")
.withEnv("POSTGRES_USER", "postgres")
.withEnv("POSTGRES_PASSWORD", "postgres")
.withExposedPorts(5432);
@@ -157,7 +157,7 @@ public class CdcSourceDatabasesIntegrationTest {
@Test
@Disabled
public void mongodb() {
GenericContainer mongodb = new GenericContainer("debezium/example-mongodb:1.3")
GenericContainer mongodb = new GenericContainer("debezium/example-mongodb:1.7.1.Final")
.withEnv("MONGODB_USER", "debezium")
.withEnv("MONGODB_PASSWORD", "dbz")
.withExposedPorts(27017);
@@ -169,7 +169,7 @@ public class CdcSourceDatabasesIntegrationTest {
"--cdc.config.mongodb.name=dbserver1",
"--cdc.config.mongodb.user=debezium",
"--cdc.config.mongodb.password=dbz",
"--cdc.config.database.whitelist=inventory")) {
"--cdc.config.collection.include.list=inventory[.]*")) {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
// Using local region here
List<Message<?>> messages = receiveAll(outputDestination);

2
applications/source/cdc-debezium-source/src/test/resources/json/mysql_insert_inventory_products_106.json
View File

@@ -22,7 +22,7 @@
"thread": null,
"query": null
},
"op": "c",
"op": "r",
"ts_ms": "${json-unit.ignore}",
"transaction": null
}

1
applications/source/mqtt-source/README.adoc
View File

@@ -24,6 +24,7 @@ $$keep-alive-interval$$:: $$the ping interval in seconds.$$ *($$Integer$$, defau
$$password$$:: $$the password to use when connecting to the broker.$$ *($$String$$, default: `$$guest$$`)*
$$persistence$$:: $$'memory' or 'file'.$$ *($$String$$, default: `$$memory$$`)*
$$persistence-directory$$:: $$Persistence directory.$$ *($$String$$, default: `$$/tmp/paho$$`)*
$$ssl-properties$$:: $$MQTT Client SSL properties.$$ *($$Map<String, String>$$, default: `$$<none>$$`)*
$$url$$:: $$location of the mqtt broker(s) (comma-delimited list).$$ *($$String[]$$, default: `$$[tcp://localhost:1883]$$`)*
$$username$$:: $$the username to use when connecting to the broker.$$ *($$String$$, default: `$$guest$$`)*

45
functions/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.3.0.Final</version.debezium>
<version.debezium>1.7.1.Final</version.debezium>
</properties>
<dependencies>
@@ -38,24 +38,24 @@
<artifactId>debezium-connector-mysql</artifactId>
<exclusions>
<exclusion>
<artifactId>mysql-connector-java</artifactId>
<groupId>mysql</groupId>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
<version>${version.debezium}</version>
</dependency>
<!-- <dependency>-->
<!-- <groupId>io.debezium</groupId>-->
<!-- <artifactId>debezium-connector-mongodb</artifactId>-->
<!-- <exclusions>-->
<!-- <exclusion>-->
<!-- <artifactId>slf4j-log4j12</artifactId>-->
<!-- <groupId>org.slf4j</groupId>-->
<!-- </exclusion>-->
<!-- </exclusions>-->
<!-- <version>${version.debezium}</version>-->
<!-- </dependency>-->
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-mongodb</artifactId>
<exclusions>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
<version>${version.debezium}</version>
</dependency>
<dependency>
<groupId>io.debezium</groupId>
@@ -91,23 +91,6 @@
<version>${version.debezium}</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>8.0.13</version>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>mysql</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jdbc</artifactId>

2
functions/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.3"))
new GenericContainer<>(DockerImageName.parse("debezium/example-mysql:1.7.1.Final"))
.withEnv("MYSQL_ROOT_PASSWORD", "debezium")
.withEnv("MYSQL_USER", "mysqluser")
.withEnv("MYSQL_PASSWORD", "mysqlpw")

42
functions/common/cdc-debezium-common/pom.xml
View File

@@ -15,12 +15,7 @@
<description>Change Data Capture (CDC) Debezium Common</description>
<properties>
<version.debezium>1.3.1.Final</version.debezium>
<!-- Note: postgresql version MUST match the version used by the Debezium postgres connector -->
<postgresql.version>42.2.5</postgresql.version>
<spring-jdbc.version>5.2.1.RELEASE</spring-jdbc.version>
<!-- Debezium requires a particular mysql-connector version -->
<mysql-connector-java>5.1.46</mysql-connector-java>
<version.debezium>1.7.1.Final</version.debezium>
</properties>
<dependencies>
@@ -39,29 +34,17 @@
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-mysql</artifactId>
<exclusions>
<exclusion>
<artifactId>mysql-connector-java</artifactId>
<groupId>mysql</groupId>
</exclusion>
</exclusions>
<optional>true</optional>
<version>${version.debezium}</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<optional>true</optional>
<version>${mysql-connector-java}</version>
</dependency>
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-postgres</artifactId>
<optional>true</optional>
<version>${version.debezium}</version>
</dependency>
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-mongodb</artifactId>
@@ -87,16 +70,6 @@
<version>${project.version}</version>
</dependency>
<!-- -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-json</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-ip</artifactId>
@@ -107,17 +80,6 @@
<version>${revision}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

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

@@ -8,6 +8,7 @@ package org.springframework.cloud.fn.common.cdc;
import java.io.IOException;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -460,6 +461,31 @@ 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() {
@@ -912,9 +938,23 @@ public final class EmbeddedEngine implements DebeziumEngine<SourceRecord> {
}
@Override
public synchronized void markBatchFinished() {
public synchronized void markBatchFinished() throws InterruptedException {
maybeFlush(offsetWriter, offsetCommitPolicy, commitTimeout, task);
}
@Override
public synchronized void markProcessed(SourceRecord record, DebeziumEngine.Offsets sourceOffsets) throws InterruptedException {
SourceRecordOffsets offsets = (SourceRecordOffsets) sourceOffsets;
SourceRecord recordWithUpdatedOffsets = new SourceRecord(record.sourcePartition(), offsets.getOffsets(), record.topic(),
record.kafkaPartition(), record.keySchema(), record.key(), record.valueSchema(), record.value(),
record.timestamp(), record.headers());
markProcessed(recordWithUpdatedOffsets);
}
@Override
public DebeziumEngine.Offsets buildOffsets() {
return new SourceRecordOffsets();
}
};
}

1
functions/consumer/mqtt-consumer/src/test/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerTests.java
View File

@@ -35,7 +35,6 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;

48
functions/supplier/cdc-debezium-supplier/pom.xml
View File

@@ -14,8 +14,7 @@
<description>CDC Debezium Suppliers</description>
<properties>
<version.debezium>1.3.1.Final</version.debezium>
<mysql.version>8.0.13</mysql.version>
<version.debezium>1.7.1.Final</version.debezium>
</properties>
<dependencies>
@@ -24,25 +23,17 @@
<artifactId>cdc-debezium-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>${mysql.version}</version>
</dependency>
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-mysql</artifactId>
<exclusions>
<exclusion>
<artifactId>mysql-connector-java</artifactId>
<groupId>mysql</groupId>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
<version>${version.debezium}</version>
</dependency>
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-mongodb</artifactId>
@@ -87,38 +78,5 @@
</exclusions>
<version>${version.debezium}</version>
</dependency>
<!-- -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

7
functions/supplier/mqtt-supplier/src/test/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierTests.java
View File

@@ -16,8 +16,6 @@
package org.springframework.cloud.fn.supplier.mqtt;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Properties;
import java.util.function.Supplier;
@@ -27,6 +25,8 @@ import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.GenericContainer;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -40,8 +40,7 @@ import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for Mqtt Supplier.