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

Refactored debezium tests to use TestContainers

Browse Source
This commit is contained in:
David Turanski
2020-10-28 16:32:28 -04:00
parent c770cfcf7f
commit 5e0f24a8ee
19 changed files with 525 additions and 477 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
applications/sink/tcp-sink/src/test/java/org/springframework/cloud/stream/app/tcp/sink/TcpSinkTests.java → applications/sink/tcp-sink/src/test/java/org/springframework/cloud/stream/app/sink/tcp/TcpSinkTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2020 the original author or authors.
* Copyright 2020-2020 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.
@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.app.tcp.sink;
package org.springframework.cloud.stream.app.sink.tcp;
import java.io.IOException;
import java.net.ServerSocket;

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

@@ -1,304 +1,202 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>cdc-debezium-source</artifactId>
<version>3.0.0-SNAPSHOT</version>
<name>cdc-debezium-source</name>
<description>CDC Debezium source apps</description>
<packaging>jar</packaging>
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.cloud.stream.app</groupId>
<artifactId>stream-applications-core</artifactId>
<version>3.0.0-SNAPSHOT</version>
<relativePath/>
</parent>
<parent>
<groupId>org.springframework.cloud.stream.app</groupId>
<artifactId>stream-applications-core</artifactId>
<version>3.1.0-SNAPSHOT</version>
<relativePath/>
</parent>
<properties>
<json-unit.version>1.25.1</json-unit.version>
</properties>
<properties>
<json-unit.version>1.25.1</json-unit.version>
</properties>
<dependencies>
<dependency>
<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>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>function-test-support</artifactId>
<version>${java-functions.version}</version>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jdbc</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>net.javacrumbs.json-unit</groupId>
<artifactId>json-unit</artifactId>
<version>${json-unit.version}</version>
<scope>test</scope>
</dependency>
<artifactId>cdc-debezium-source</artifactId>
<name>cdc-debezium-source</name>
<description>CDC Debezium source apps</description>
<packaging>jar</packaging>
<dependencies>
<dependency>
<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>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>function-test-support</artifactId>
<version>${java-functions.version}</version>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.cloud.stream.app</groupId>
<artifactId>stream-applications-test-support</artifactId>
<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>
<scope>test</scope>
</dependency>
<dependency>
<groupId>net.javacrumbs.json-unit</groupId>
<artifactId>json-unit</artifactId>
<version>${json-unit.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-params</artifactId>
<scope>test</scope>
</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.cloud.fn</groupId>
<artifactId>cdc-debezium-boot-starter</artifactId>
<version>1.1.0-SNAPSHOT</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dataflow-apps-docs-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dataflow-apps-generator-plugin</artifactId>
<configuration>
<application>
<name>cdc-debezium</name>
<type>source</type>
<version>${project.version}</version>
<configClass>org.springframework.cloud.fn.supplier.cdc.CdcSupplierConfiguration.class</configClass>
<functionDefinition>cdcSupplier</functionDefinition>
<maven>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>cdc-debezium-supplier</artifactId>
</dependency>
</dependencies>
</maven>
</application>
</configuration>
</plugin>
<plugin>
<groupId>io.fabric8</groupId>
<artifactId>docker-maven-plugin</artifactId>
<version>0.33.0</version>
<configuration>
<skip>${skipTests}</skip>
<images>
<!-- <image>-->
<!-- <alias>test-mongodb2</alias>-->
<!-- <name>%a/example-mongodb:${project.version}</name>-->
<!-- <build>-->
<!-- <dockerFileDir>${project.basedir}/src/test/docker/mongodb</dockerFileDir>-->
<!-- <filter>@</filter>-->
<!-- </build>-->
<!-- <run>-->
<!-- <hostname>localhost</hostname>-->
<!-- <env>-->
<!-- <MONGODB_USER>debezium</MONGODB_USER>-->
<!-- <MONGODB_PASSWORD>dbz</MONGODB_PASSWORD>-->
<!-- </env>-->
<!-- <ports>-->
<!-- <port>27017:27017</port>-->
<!-- </ports>-->
<!-- <wait>-->
<!-- <log>port: 3306</log>-->
<!-- <time>300000</time>-->
<!-- </wait>-->
<!-- </run>-->
<!-- </image>-->
<!--<image>-->
<!--<alias>test-mongodb</alias>-->
<!--<name>debezium/example-mongodb:1.0</name>-->
<!--<run>-->
<!--<hostname>localhost</hostname>-->
<!--<env>-->
<!--<MONGODB_USER>debezium</MONGODB_USER>-->
<!--<MONGODB_PASSWORD>dbz</MONGODB_PASSWORD>-->
<!--</env>-->
<!--<ports>-->
<!--<port>27017:27017</port>-->
<!--</ports>-->
<!--<wait>-->
<!--<log>port: 3306</log>-->
<!--<time>300000</time>-->
<!--</wait>-->
<!--</run>-->
<!--</image>-->
<image>
<alias>mysql</alias>
<name>debezium/example-mysql:1.0</name>
<run>
<env>
<MYSQL_ROOT_PASSWORD>debezium</MYSQL_ROOT_PASSWORD>
<MYSQL_USER>mysqluser</MYSQL_USER>
<MYSQL_PASSWORD>mysqlpw</MYSQL_PASSWORD>
</env>
<ports>
<port>3306:3306</port>
</ports>
<wait>
<log>port: 3306</log>
<time>30000</time>
</wait>
</run>
</image>
<image>
<alias>postgres</alias>
<name>debezium/example-postgres:1.0</name>
<run>
<env>
<POSTGRES_USER>postgres</POSTGRES_USER>
<POSTGRES_PASSWORD>postgres</POSTGRES_PASSWORD>
</env>
<ports>
<port>5432:5432</port>
</ports>
<wait>
<log>PostgreSQL init process complete</log>
<time>30000</time>
</wait>
</run>
</image>
<image>
<external>
<type>properties</type>
<prefix>docker</prefix>
<mode>override</mode>
</external>
<name>%a/sqlserver-example:${project.version}</name>
<build>
<dockerFileDir>${project.basedir}/src/test/docker/sqlserver</dockerFileDir>
<filter>@</filter>
</build>
<run>
<env>
<ACCEPT_EULA>Y</ACCEPT_EULA>
<MSSQL_PID>Standard</MSSQL_PID>
<SA_PASSWORD>Password!</SA_PASSWORD>
<MSSQL_AGENT_ENABLED>true</MSSQL_AGENT_ENABLED>
</env>
<ports>
<port>1433:1433</port>
</ports>
<wait>
<log>1 rows affected</log>
<time>40000</time>
</wait>
</run>
</image>
</images>
</configuration>
<executions>
<execution>
<id>start</id>
<!--<phase>pre-integration-test</phase>-->
<phase>generate-test-resources</phase>
<goals>
<goal>build</goal>
<goal>start</goal>
</goals>
</execution>
<execution>
<id>stop</id>
<phase>post-integration-test</phase>
<goals>
<goal>stop</goal>
</goals>
</execution>
<execution>
<id>stop-pre</id>
<phase>clean</phase>
<goals>
<goal>stop</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<snapshots>
<enabled>true</enabled>
</snapshots>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
</repository>
<repository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
</repository>
<repository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/release</url>
</repository>
<repository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-libs-release</id>
<name>Spring Libs Release</name>
<url>https://repo.spring.io/libs-release</url>
</repository>
<repository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-milestone-release</id>
<name>Spring Milestone Release</name>
<url>https://repo.spring.io/libs-milestone</url>
</repository>
</repositories>
<pluginRepositories>
<pluginRepository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/libs-release</url>
</pluginRepository>
<pluginRepository>
<snapshots>
<enabled>true</enabled>
</snapshots>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
</pluginRepository>
<pluginRepository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
</pluginRepository>
</pluginRepositories>
<build>
<plugins>
<plugin>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dataflow-apps-docs-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dataflow-apps-generator-plugin</artifactId>
<configuration>
<application>
<name>cdc-debezium</name>
<type>source</type>
<version>${project.version}</version>
<configClass>org.springframework.cloud.fn.supplier.cdc.CdcSupplierConfiguration.class
</configClass>
<functionDefinition>cdcSupplier</functionDefinition>
<maven>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>cdc-debezium-supplier</artifactId>
</dependency>
</dependencies>
</maven>
</application>
</configuration>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<snapshots>
<enabled>true</enabled>
</snapshots>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
</repository>
<repository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
</repository>
<repository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/release</url>
</repository>
<repository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-libs-release</id>
<name>Spring Libs Release</name>
<url>https://repo.spring.io/libs-release</url>
</repository>
<repository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-milestone-release</id>
<name>Spring Milestone Release</name>
<url>https://repo.spring.io/libs-milestone</url>
</repository>
</repositories>
<pluginRepositories>
<pluginRepository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/libs-release</url>
</pluginRepository>
<pluginRepository>
<snapshots>
<enabled>true</enabled>
</snapshots>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
</pluginRepository>
<pluginRepository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
</pluginRepository>
</pluginRepositories>
</project>

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

@@ -19,7 +19,9 @@ package org.springframework.cloud.stream.app.source.cdc;
import java.time.Duration;
import java.util.List;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.springframework.boot.test.context.FilteredClassLoader;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
@@ -28,7 +30,6 @@ import org.springframework.cloud.fn.common.cdc.CdcCommonProperties;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ApplicationContext;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.support.KafkaNull;
import org.springframework.messaging.Message;
import org.springframework.test.jdbc.JdbcTestUtils;
@@ -40,19 +41,17 @@ import static org.springframework.cloud.stream.app.source.cdc.CdcTestUtils.recei
/**
* @author Christian Tzolov
* @author David Turanski
*/
public class CdcDeleteHandlingIntegrationTest {
private final JdbcTemplate jdbcTemplate = CdcTestUtils.jdbcTemplate(
"com.mysql.cj.jdbc.Driver",
"jdbc:mysql://localhost:3306/inventory",
"root", "debezium");
@Testcontainers
public class CdcDeleteHandlingIntegrationTest extends CdcTestSupport {
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withUserConfiguration(
TestChannelBinderConfiguration.getCompleteConfiguration(TestCdcSourceApplication.class))
.withPropertyValues(
"spring.cloud.stream.function.definition=cdcSupplier",
"spring.cloud.function.definition=cdcSupplier",
"cdc.name=my-sql-connector",
"cdc.schema=false",
"cdc.flattering.enabled=true",
@@ -61,60 +60,45 @@ public class CdcDeleteHandlingIntegrationTest {
"cdc.config.database.user=debezium",
"cdc.config.database.password=dbz",
"cdc.config.database.hostname=localhost",
"cdc.config.database.port=3306",
"cdc.config.database.server.id=85744",
"cdc.config.database.port=" + MAPPED_PORT,
// "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 handleRecordDeletionTest() {
contextRunner.withPropertyValues("cdc.flattering.deleteHandlingMode=none", "cdc.flattering.dropTombstones=true")
.run(consumer);
contextRunner.withPropertyValues("cdc.flattering.deleteHandlingMode=none", "cdc.flattering.dropTombstones=true")
.withClassLoader(new FilteredClassLoader(KafkaNull.class)) // Remove Kafka from the classpath
@ParameterizedTest
@ValueSource(strings = {
"cdc.flattering.deleteHandlingMode=none,cdc.flattering.dropTombstones=true",
"cdc.flattering.deleteHandlingMode=none,cdc.flattering.dropTombstones=false",
"cdc.flattering.deleteHandlingMode=drop,cdc.flattering.dropTombstones=true",
"cdc.flattering.deleteHandlingMode=drop,cdc.flattering.dropTombstones=false",
"cdc.flattering.deleteHandlingMode=rewrite,cdc.flattering.dropTombstones=true",
"cdc.flattering.deleteHandlingMode=rewrite,cdc.flattering.dropTombstones=false"
})
public void handleRecordDeletions(String properties) {
contextRunner.withPropertyValues(properties.split(","))
.withClassLoader(new FilteredClassLoader(KafkaNull.class)) // Remove Kafka from the
.run(consumer);
contextRunner.withPropertyValues("cdc.flattering.deleteHandlingMode=none", "cdc.flattering.dropTombstones=false")
.run(consumer);
contextRunner.withPropertyValues("cdc.flattering.deleteHandlingMode=none", "cdc.flattering.dropTombstones=false")
.withClassLoader(new FilteredClassLoader(KafkaNull.class)) // Remove Kafka from the classpath
contextRunner.withPropertyValues(properties.split(","))
.run(consumer);
}
contextRunner.withPropertyValues("cdc.flattering.deleteHandlingMode=drop", "cdc.flattering.dropTombstones=true")
.run(consumer);
contextRunner.withPropertyValues("cdc.flattering.deleteHandlingMode=drop", "cdc.flattering.dropTombstones=true")
.withClassLoader(new FilteredClassLoader(KafkaNull.class)) // Remove Kafka from the classpath
.run(consumer);
contextRunner.withPropertyValues("cdc.flattering.deleteHandlingMode=drop", "cdc.flattering.dropTombstones=false")
.run(consumer);
contextRunner.withPropertyValues("cdc.flattering.deleteHandlingMode=drop", "cdc.flattering.dropTombstones=false")
.withClassLoader(new FilteredClassLoader(KafkaNull.class)) // Remove Kafka from the classpath
.run(consumer);
contextRunner.withPropertyValues("cdc.flattering.deleteHandlingMode=rewrite", "cdc.flattering.dropTombstones=true")
.run(consumer);
contextRunner.withPropertyValues("cdc.flattering.deleteHandlingMode=rewrite", "cdc.flattering.dropTombstones=true")
.withClassLoader(new FilteredClassLoader(KafkaNull.class)) // Remove Kafka from the classpath
.run(consumer);
contextRunner.withPropertyValues("cdc.flattering.deleteHandlingMode=rewrite", "cdc.flattering.dropTombstones=false")
.run(consumer);
contextRunner.withPropertyValues("cdc.flattering.deleteHandlingMode=rewrite", "cdc.flattering.dropTombstones=false")
.withClassLoader(new FilteredClassLoader(KafkaNull.class)) // Remove Kafka from the classpath
.run(consumer);
private String toString(Object object) {
return new String((byte[]) object);
}
final ContextConsumer<? super ApplicationContext> consumer = context -> {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
CdcCommonProperties props = context.getBean(CdcCommonProperties.class);
boolean isKafkaPresent = ClassUtils.isPresent(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL, context.getClassLoader());
boolean isKafkaPresent = ClassUtils.isPresent(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL,
context.getClassLoader());
CdcCommonProperties.DeleteHandlingMode deleteHandlingMode = props.getFlattering().getDeleteHandlingMode();
boolean isDropTombstones = props.getFlattering().isDropTombstones();
jdbcTemplate.update("insert into `customers`(`first_name`,`last_name`,`email`) VALUES('Test666', 'Test666', 'Test666@spring.org')");
jdbcTemplate.update(
"insert into `customers`(`first_name`,`last_name`,`email`) VALUES('Test666', 'Test666', 'Test666@spring.org')");
String newRecordId = jdbcTemplate.query("select * from `customers` where `first_name` = ?",
(rs, rowNum) -> rs.getString("id"), "Test666").iterator().next();
@@ -140,23 +124,18 @@ public class CdcDeleteHandlingIntegrationTest {
}
if (!isDropTombstones && isKafkaPresent) {
received = outputDestination.receive(Duration.ofSeconds(10).toMillis());
received = outputDestination.receive(Duration.ofSeconds(10000).toMillis());
assertThat(received).isNotNull();
//Tombstones event should have KafkaNull payload
// Tombstones event should have KafkaNull payload
assertThat(received.getPayload().getClass().getCanonicalName())
.isEqualTo(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL);
String key = (String) received.getHeaders().get("cdc_key");
//Tombstones event should carry the deleted record id in the cdc_key header
// Tombstones event should carry the deleted record id in the cdc_key header
assertThat(key).isEqualTo("{\"id\":" + newRecordId + "}");
}
received = outputDestination.receive(Duration.ofSeconds(10).toMillis());
assertThat(received).isNull();
};
private String toString(Object object) {
return new String((byte[]) object);
}
}

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

@@ -27,7 +27,6 @@ import org.springframework.cloud.fn.common.cdc.CdcCommonProperties;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ApplicationContext;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.support.KafkaNull;
import org.springframework.messaging.Message;
import org.springframework.test.jdbc.JdbcTestUtils;
@@ -42,13 +41,9 @@ import static org.springframework.cloud.stream.app.source.cdc.CdcTestUtils.resou
/**
* @author Christian Tzolov
* @author David Turanski
*/
public class CdcFlatteringIntegrationTest {
private final JdbcTemplate jdbcTemplate = CdcTestUtils.jdbcTemplate(
"com.mysql.cj.jdbc.Driver",
"jdbc:mysql://localhost:3306/inventory",
"root", "debezium");
public class CdcFlatteringIntegrationTest extends CdcTestSupport {
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withUserConfiguration(
@@ -57,18 +52,15 @@ public class CdcFlatteringIntegrationTest {
"spring.cloud.stream.function.definition=cdcSupplier",
"cdc.name=my-sql-connector",
"cdc.schema=false",
"cdc.stream.header.offset=false",
"cdc.connector=mysql",
"cdc.config.database.user=debezium",
"cdc.config.database.password=dbz",
"cdc.config.database.hostname=localhost",
"cdc.config.database.port=3306",
"cdc.config.database.server.id=85744",
"cdc.config.database.port=" + MAPPED_PORT,
// "cdc.config.database.server.id=85744",
"cdc.config.database.server.name=my-app-connector",
"cdc.config.database.history=io.debezium.relational.history.MemoryDatabaseHistory"
);
"cdc.config.database.history=io.debezium.relational.history.MemoryDatabaseHistory");
@Test
public void noFlatteredResponseNoKafka() {
@@ -85,7 +77,8 @@ public class CdcFlatteringIntegrationTest {
final ContextConsumer<? super ApplicationContext> noFlatteringTest = context -> {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
boolean isKafkaPresent = ClassUtils.isPresent(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL, context.getClassLoader());
boolean isKafkaPresent = ClassUtils.isPresent(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL,
context.getClassLoader());
List<Message<?>> messages = receiveAll(outputDestination);
assertThat(messages).hasSizeGreaterThanOrEqualTo(52);
@@ -169,7 +162,8 @@ public class CdcFlatteringIntegrationTest {
final ContextConsumer<? super ApplicationContext> flatteringTest = context -> {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
boolean isKafkaPresent = ClassUtils.isPresent(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL, context.getClassLoader());
boolean isKafkaPresent = ClassUtils.isPresent(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL,
context.getClassLoader());
List<Message<?>> messages = receiveAll(outputDestination);
assertThat(messages).hasSizeGreaterThanOrEqualTo(52);
@@ -222,7 +216,8 @@ public class CdcFlatteringIntegrationTest {
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 + "}", messages.get(1).getHeaders().get("cdc_key"));
if (!StringUtils.isEmpty(flatteringProps.getAddHeaders()) && flatteringProps.getAddHeaders().contains("op")) {
if (!StringUtils.isEmpty(flatteringProps.getAddHeaders())
&& flatteringProps.getAddHeaders().contains("op")) {
assertThat(messages.get(2).getHeaders().get("__op")).isEqualTo("d");
}
}

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

@@ -16,38 +16,45 @@
package org.springframework.cloud.stream.app.source.cdc;
import java.time.Duration;
import java.util.List;
import org.junit.jupiter.api.Test;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.images.builder.ImageFromDockerfile;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.stream.app.test.integration.LogMatcher;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.springframework.cloud.stream.app.source.cdc.CdcTestUtils.receiveAll;
/**
* @author Christian Tzolov
* @author David Turanski
*/
public class CdcSourceDatabasesIntegrationTest<b> {
public class CdcSourceDatabasesIntegrationTest extends CdcTestSupport {
private final SpringApplicationBuilder applicationBuilder = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(TestCdcSourceApplication.class))
.web(WebApplicationType.NONE)
.properties("spring.cloud.stream.function.definition=cdcSupplier",
"cdc.name=my-sql-connector",
"cdc.flattering.dropTombstones=false",
"cdc.schema=false",
"cdc.flattering.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");
.web(WebApplicationType.NONE)
.properties("spring.cloud.stream.function.definition=cdcSupplier",
"cdc.name=my-sql-connector",
"cdc.flattering.dropTombstones=false",
"cdc.schema=false",
"cdc.flattering.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() {
@@ -56,7 +63,7 @@ public class CdcSourceDatabasesIntegrationTest<b> {
"--cdc.config.database.user=debezium",
"--cdc.config.database.password=dbz",
"--cdc.config.database.hostname=localhost",
"--cdc.config.database.port=3306")) {
"--cdc.config.database.port=" + MAPPED_PORT)) {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
// Using local region here
List<Message<?>> messages = receiveAll(outputDestination);
@@ -67,17 +74,32 @@ public class CdcSourceDatabasesIntegrationTest<b> {
@Test
public void sqlServer() {
LogMatcher logMatcher = LogMatcher.contains("(1 rows affected)").times(26);
GenericContainer sqlServer = new GenericContainer(new ImageFromDockerfile()
.withFileFromClasspath("Dockerfile", "sqlserver/Dockerfile")
.withFileFromClasspath("import-data.sh", "sqlserver/import-data.sh")
.withFileFromClasspath("inventory.sql", "sqlserver/inventory.sql")
.withFileFromClasspath("entrypoint.sh", "sqlserver/entrypoint.sh"))
.withEnv("ACCEPT_EULA", "Y")
.withEnv("MSSQL_PID", "Standard")
.withEnv("SA_PASSWORD", "Password!")
.withEnv("MSSQL_AGENT_ENABLED", "true")
.withLogConsumer(logMatcher)
.withLogConsumer(new Slf4jLogConsumer(LoggerFactory.getLogger("sqlServer")))
.withExposedPorts(1433);
sqlServer.start();
assertThat(sqlServer.isRunning());
await().atMost(Duration.ofSeconds(60)).until(logMatcher.matches());
try (ConfigurableApplicationContext context = applicationBuilder
.run("--cdc.connector=sqlserver",
//"--cdc.config.database.user=Standard",
// "--cdc.config.database.user=Standard",
"--cdc.config.database.user=sa",
"--cdc.config.database.password=Password!",
"--cdc.config.database.dbname=testDB",
"--cdc.config.database.hostname=localhost",
"--cdc.config.database.port=1433"
)) {
"--cdc.config.database.port=" + sqlServer.getMappedPort(1433))) {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
// Using local region here
List<Message<?>> messages = receiveAll(outputDestination);
@@ -88,27 +110,40 @@ public class CdcSourceDatabasesIntegrationTest<b> {
@Test
public void postgres() {
GenericContainer postgres = new GenericContainer("debezium/example-postgres:1.0")
.withEnv("POSTGRES_USER", "postgres")
.withEnv("POSTGRES_PASSWORD", "postgres")
.withExposedPorts(5432);
postgres.start();
try (ConfigurableApplicationContext context = applicationBuilder
.run("--cdc.connector=postgres",
"--cdc.config.database.user=postgres",
"--cdc.config.database.password=postgres",
"--cdc.config.slot.name=debezium",
"--cdc.config.database.dbname=postgres",
"--cdc.config.database.hostname=localhost",
"--cdc.config.database.port=5432")) {
"--cdc.config.database.port=" + postgres.getMappedPort(5432))) {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
// Using local region here
List<Message<?>> messages = receiveAll(outputDestination);
assertThat(messages).isNotNull();
assertThat(messages).hasSize(5786);
}
postgres.stop();
}
//@Test
// @Test
public void mongodb() {
GenericContainer mongodb = new GenericContainer("debezium/example-mongodb:1.0")
.withEnv("MONGODB_USER", "debezium")
.withEnv("MONGODB_PASSWORD", "dbz")
.withExposedPorts(27017);
mongodb.start();
try (ConfigurableApplicationContext context = applicationBuilder
.run("--cdc.connector=mongodb",
"--cdc.config.tasks.max=1",
"--cdc.config.mongodb.hosts=rs0/localhost:27017",
"--cdc.config.mongodb.hosts=rs0/localhost:" + mongodb.getMappedPort(27017),
"--cdc.config.mongodb.name=dbserver1",
"--cdc.config.mongodb.user=debezium",
"--cdc.config.mongodb.password=dbz",
@@ -119,5 +154,6 @@ public class CdcSourceDatabasesIntegrationTest<b> {
assertThat(messages).isNotNull();
assertThat(messages).hasSize(666);
}
mongodb.stop();
}
}

55
applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcTestSupport.java Normal file
View File

@@ -0,0 +1,55 @@
/*
* Copyright 2020-2020 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.app.source.cdc;
import org.junit.jupiter.api.BeforeAll;
import org.testcontainers.containers.GenericContainer;
import org.springframework.jdbc.core.JdbcTemplate;
/**
* @author David Turanski
*/
public abstract class CdcTestSupport {
static final String DATABASE_NAME = "inventory";
static String MAPPED_PORT;
static GenericContainer debeziumMySQL = new GenericContainer<>("debezium/example-mysql:1.0")
.withEnv("MYSQL_ROOT_PASSWORD", "debezium")
.withEnv("MYSQL_USER", "mysqluser")
.withEnv("MYSQL_PASSWORD", "mysqlpw")
// .withLogConsumer(new Slf4jLogConsumer(LoggerFactory.getLogger("mysql")))
.withExposedPorts(3306);
static {
debeziumMySQL.start();
}
static JdbcTemplate jdbcTemplate;
@BeforeAll
static void setup() {
MAPPED_PORT = String.valueOf(debeziumMySQL.getMappedPort(3306));
jdbcTemplate = CdcTestUtils.jdbcTemplate(
"com.mysql.cj.jdbc.Driver",
"jdbc:mysql://localhost:" + MAPPED_PORT + "/" + DATABASE_NAME,
"root",
"debezium");
}
}

30
applications/source/cdc-debezium-source/src/test/resources/logback-test.xml Normal file
View File

@@ -0,0 +1,30 @@
<!--
~ Copyright 2020-2020 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.
~ You may obtain a copy of the License at
~
~ https://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing, software
~ distributed under the License is distributed on an "AS IS" BASIS,
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
~ See the License for the specific language governing permissions and
~ limitations under the License.
-->
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n</pattern>
</encoder>
</appender>
<root level="info">
<appender-ref ref="STDOUT"/>
</root>
<logger name="org.testcontainers" level="INFO"/>
<logger name="com.github.dockerjava" level="WARN"/>
</configuration>

2
applications/source/cdc-debezium-source/src/test/docker/sqlserver/Dockerfile → applications/source/cdc-debezium-source/src/test/resources/sqlserver/Dockerfile
View File

@@ -8,6 +8,6 @@ WORKDIR /usr/src/data
COPY . /usr/src/data
# Grant permissions for the import-data script to be executable
RUN chmod +x /usr/src/data/import-data.sh
RUN chmod a+x /usr/src/data/*.sh
CMD /bin/bash ./entrypoint.sh

0
applications/source/cdc-debezium-source/src/test/docker/sqlserver/entrypoint.sh → applications/source/cdc-debezium-source/src/test/resources/sqlserver/entrypoint.sh
View File

0
applications/source/cdc-debezium-source/src/test/docker/sqlserver/import-data.sh → applications/source/cdc-debezium-source/src/test/resources/sqlserver/import-data.sh
View File

0
applications/source/cdc-debezium-source/src/test/docker/sqlserver/inventory.sql → applications/source/cdc-debezium-source/src/test/resources/sqlserver/inventory.sql
View File

4
applications/stream-applications-core/stream-applications-test-support/pom.xml
View File

@@ -28,28 +28,24 @@
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
<version>${test-containers.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>junit-jupiter</artifactId>
<version>${test-containers.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>kafka</artifactId>
<version>${test-containers.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>rabbitmq</artifactId>
<version>${test-containers.version}</version>
<optional>true</optional>
</dependency>

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

@@ -45,17 +45,17 @@
<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>
@@ -97,68 +97,28 @@
<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>
<scope>test</scope>
</dependency>
</dependencies>
<dependency>
<groupId>com.zaxxer</groupId>
<artifactId>HikariCP</artifactId>
<scope>test</scope>
</dependency>
<build>
<plugins>
<!-- <plugin>-->
<!-- <groupId>io.fabric8</groupId>-->
<!-- <artifactId>docker-maven-plugin</artifactId>-->
<!-- <version>0.33.0</version>-->
<!-- <configuration>-->
<!-- <images>-->
<!-- <image>-->
<!-- <alias>mysql</alias>-->
<!-- <name>debezium/example-mysql:1.0</name>-->
<!-- <run>-->
<!-- <env>-->
<!-- <MYSQL_ROOT_PASSWORD>debezium</MYSQL_ROOT_PASSWORD>-->
<!-- <MYSQL_USER>mysqluser</MYSQL_USER>-->
<!-- <MYSQL_PASSWORD>mysqlpw</MYSQL_PASSWORD>-->
<!-- </env>-->
<!-- <ports>-->
<!-- <port>3306:3306</port>-->
<!-- </ports>-->
<!-- <wait>-->
<!-- <log>port: 3306</log>-->
<!-- <time>30000</time>-->
<!-- </wait>-->
<!-- </run>-->
<!-- </image>-->
<!-- </images>-->
<!-- </configuration>-->
<!-- <executions>-->
<!-- <execution>-->
<!-- <id>start</id>-->
<!-- &lt;!&ndash;<phase>pre-integration-test</phase>&ndash;&gt;-->
<!-- <phase>generate-test-resources</phase>-->
<!-- <goals>-->
<!-- <goal>build</goal>-->
<!-- <goal>start</goal>-->
<!-- </goals>-->
<!-- </execution>-->
<!-- <execution>-->
<!-- <id>stop</id>-->
<!-- <phase>post-integration-test</phase>-->
<!-- <goals>-->
<!-- <goal>stop</goal>-->
<!-- </goals>-->
<!-- </execution>-->
<!-- <execution>-->
<!-- <id>stop-pre</id>-->
<!-- <phase>clean</phase>-->
<!-- <goals>-->
<!-- <goal>stop</goal>-->
<!-- </goals>-->
<!-- </execution>-->
<!-- </executions>-->
<!-- </plugin>-->
</plugins>
</build>
</dependencies>
</project>

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

@@ -16,29 +16,58 @@
package org.springframework.cloud.fn.common.cdc;
import org.junit.Test;
import java.time.Duration;
import com.zaxxer.hikari.HikariDataSource;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DriverManagerDataSource;
import org.springframework.test.jdbc.JdbcTestUtils;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
/**
* @author Christian Tzolov
* @author David Turanski
*/
@Testcontainers
public class CdcBootStarterIntegrationTest {
private final JdbcTemplate jdbcTemplate = jdbcTemplate(
"com.mysql.cj.jdbc.Driver",
"jdbc:mysql://localhost:3306/inventory",
"root",
"debezium");
private static final String DATABASE_NAME = "inventory";
private static String MAPPED_PORT;
@Container
static GenericContainer debeziumMySQL = new GenericContainer<>(DockerImageName.parse("debezium/example-mysql:1.0"))
.withEnv("MYSQL_ROOT_PASSWORD", "debezium")
.withEnv("MYSQL_USER", "mysqluser")
.withEnv("MYSQL_PASSWORD", "mysqlpw")
// .withLogConsumer(new Slf4jLogConsumer(LoggerFactory.getLogger("mysql")))
.withExposedPorts(3306);
private static JdbcTemplate jdbcTemplate;
@BeforeAll
static void setup() {
MAPPED_PORT = String.valueOf(debeziumMySQL.getMappedPort(3306));
jdbcTemplate = jdbcTemplate(
"com.mysql.cj.jdbc.Driver",
"jdbc:mysql://localhost:" + MAPPED_PORT + "/" + DATABASE_NAME,
"root",
"debezium");
}
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withUserConfiguration(TestCdcApplication.class)
.withPropertyValues(
"spring.datasource.type=com.zaxxer.hikari.HikariDataSource",
"cdc.name=my-sql-connector",
"cdc.schema=false",
"cdc.flattering.enabled=true",
@@ -47,7 +76,7 @@ public class CdcBootStarterIntegrationTest {
"cdc.config.database.user=debezium",
"cdc.config.database.password=dbz",
"cdc.config.database.hostname=localhost",
"cdc.config.database.port=3306",
"cdc.config.database.port=" + MAPPED_PORT,
"cdc.config.database.server.id=85744",
"cdc.config.database.server.name=my-app-connector",
"cdc.config.database.history=io.debezium.relational.history.MemoryDatabaseHistory");
@@ -59,20 +88,21 @@ public class CdcBootStarterIntegrationTest {
"cdc.flattering.deleteHandlingMode=drop",
"cdc.flattering.dropTombstones=true")
.run(context -> {
TestCdcApplication.TestSourceRecordConsumer testConsumer =
context.getBean(TestCdcApplication.TestSourceRecordConsumer.class);
jdbcTemplate.update("insert into `customers`(`first_name`,`last_name`,`email`) VALUES('Test666', 'Test666', 'Test666@spring.org')");
TestCdcApplication.TestSourceRecordConsumer testConsumer = context
.getBean(TestCdcApplication.TestSourceRecordConsumer.class);
jdbcTemplate.update(
"insert into `customers`(`first_name`,`last_name`,`email`) VALUES('Test666', 'Test666', 'Test666@spring.org')");
JdbcTestUtils.deleteFromTableWhere(jdbcTemplate, "customers", "first_name = ?", "Test666");
Thread.sleep(10000);
assertThat(testConsumer.recordList).hasSizeGreaterThanOrEqualTo(52);
await().atMost(Duration.ofSeconds(30))
.untilAsserted(() -> assertThat(testConsumer.recordList).hasSizeGreaterThanOrEqualTo(52));
});
}
public static JdbcTemplate jdbcTemplate(String jdbcDriver, String jdbcUrl, String user, String password) {
DriverManagerDataSource dataSource = new DriverManagerDataSource();
HikariDataSource dataSource = new HikariDataSource();
dataSource.setDriverClassName(jdbcDriver);
dataSource.setUrl(jdbcUrl);
dataSource.setJdbcUrl(jdbcUrl);
dataSource.setUsername(user);
dataSource.setPassword(password);
return new JdbcTemplate(dataSource);

11
functions/common/cdc-debezium-boot-starter/src/test/java/org/springframework/cloud/fn/common/cdc/TestCdcApplication.java
View File

@@ -16,10 +16,10 @@
package org.springframework.cloud.fn.common.cdc;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -27,6 +27,7 @@ import org.apache.kafka.connect.source.SourceRecord;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.metadata.SimpleMetadataStore;
@@ -34,7 +35,7 @@ import org.springframework.integration.metadata.SimpleMetadataStore;
* @author Christian Tzolov
*/
@SpringBootConfiguration
@EnableAutoConfiguration
@EnableAutoConfiguration(exclude = { DataSourceAutoConfiguration.class })
public class TestCdcApplication {
@Bean
@@ -51,10 +52,12 @@ public class TestCdcApplication {
public static class TestSourceRecordConsumer implements Consumer<SourceRecord> {
private final Function<SourceRecord, byte[]> valueSerializer;
private final Function<SourceRecord, byte[]> keySerializer;
public Map<Object, Object> keyValue = new HashMap<>();
public List<SourceRecord> recordList = new ArrayList<>();
public List<SourceRecord> recordList = new CopyOnWriteArrayList<>();
public TestSourceRecordConsumer(Function<SourceRecord, byte[]> valueSerializer,
Function<SourceRecord, byte[]> keySerializer) {
@@ -75,5 +78,3 @@ public class TestCdcApplication {
}
}
}

30
functions/common/cdc-debezium-boot-starter/src/test/resources/logback-test.xml Normal file
View File

@@ -0,0 +1,30 @@
<!--
~ Copyright 2020-2020 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.
~ You may obtain a copy of the License at
~
~ https://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing, software
~ distributed under the License is distributed on an "AS IS" BASIS,
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
~ See the License for the specific language governing permissions and
~ limitations under the License.
-->
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n</pattern>
</encoder>
</appender>
<root level="info">
<appender-ref ref="STDOUT"/>
</root>
<logger name="org.testcontainers" level="INFO"/>
<logger name="com.github.dockerjava" level="WARN"/>
</configuration>

12
functions/spring-functions-parent/pom.xml
View File

@@ -75,6 +75,18 @@
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>

31
functions/supplier/cdc-debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/cdc/CdcSupplierConfiguration.java
View File

@@ -46,8 +46,8 @@ import org.springframework.util.ClassUtils;
import org.springframework.util.MimeTypeUtils;
/**
* CDC source that uses the Debezium Connectors to monitor and record all of the row-level changes in the databases.
* https://debezium.io/docs/connectors
* CDC source that uses the Debezium Connectors to monitor and record all of the row-level
* changes in the databases. https://debezium.io/docs/connectors
*
* @author Christian Tzolov
*/
@@ -91,6 +91,22 @@ public class CdcSupplierConfiguration implements BeanClassLoaderAware {
private EmitterProcessor<Message<?>> emitterProcessor = EmitterProcessor.create(1, false);
// @Bean
// public MessageConverter kafkaNullMessageConverter() {
// return new AbstractMessageConverter() {
// @Override
// protected Object convertToInternal(Object payload,
// MessageHeaders headers, Object conversionHint) {
// return payload;
// }
//
// @Override
// protected boolean supports(Class<?> clazz) {
// return clazz.getName().equals(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL);
// }
// };
// }
@Bean
public EmbeddedEngineExecutorService embeddedEngineExecutorService(
EmbeddedEngine.Builder embeddedEngineBuilder,
@@ -111,9 +127,11 @@ public class CdcSupplierConfiguration implements BeanClassLoaderAware {
Object cdcJsonPayload = valueSerializer.apply(sourceRecord);
// When the tombstone event is enabled, Debezium serializes the payload to null (e.g. empty payload)
// When the tombstone event is enabled, Debezium serializes the payload to null (e.g.
// empty payload)
// while the metadata information is carried through the headers (cdc_key).
// Note: Event for none flattered responses, when the cdc.config.tombstones.on.delete=true (default),
// Note: Event for none flattered responses, when the cdc.config.tombstones.on.delete=true
// (default),
// tombstones are generate by Debezium and handled by the code below.
if (cdcJsonPayload == null) {
cdcJsonPayload = this.kafkaNull;
@@ -135,7 +153,9 @@ public class CdcSupplierConfiguration implements BeanClassLoaderAware {
.withPayload(cdcJsonPayload)
.setHeader("cdc_key", new String(key))
.setHeader("cdc_topic", sourceRecord.topic())
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE);
.setHeader(MessageHeaders.CONTENT_TYPE,
(cdcJsonPayload.equals(this.kafkaNull)) ? MimeTypeUtils.TEXT_PLAIN_VALUE
: MimeTypeUtils.APPLICATION_JSON_VALUE);
if (cdcStreamingEngineProperties.getHeader().isConvertConnectHeaders()) {
// Convert the Connect Headers into Message Headers.
@@ -158,7 +178,6 @@ public class CdcSupplierConfiguration implements BeanClassLoaderAware {
}
}
sink.next(messageBuilder.build());
};

7
stream-applications-build/pom.xml
View File

@@ -74,6 +74,13 @@
<type>pom</type>
<scope>import</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers-bom</artifactId>
<version>${test-containers.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>