diff --git a/applications/sink/tcp-sink/src/test/java/org/springframework/cloud/stream/app/tcp/sink/TcpSinkTests.java b/applications/sink/tcp-sink/src/test/java/org/springframework/cloud/stream/app/sink/tcp/TcpSinkTests.java similarity index 98% rename from applications/sink/tcp-sink/src/test/java/org/springframework/cloud/stream/app/tcp/sink/TcpSinkTests.java rename to applications/sink/tcp-sink/src/test/java/org/springframework/cloud/stream/app/sink/tcp/TcpSinkTests.java index f56eb900..1b8354cd 100644 --- a/applications/sink/tcp-sink/src/test/java/org/springframework/cloud/stream/app/tcp/sink/TcpSinkTests.java +++ b/applications/sink/tcp-sink/src/test/java/org/springframework/cloud/stream/app/sink/tcp/TcpSinkTests.java @@ -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; diff --git a/applications/source/cdc-debezium-source/pom.xml b/applications/source/cdc-debezium-source/pom.xml index 2b791441..f279a841 100644 --- a/applications/source/cdc-debezium-source/pom.xml +++ b/applications/source/cdc-debezium-source/pom.xml @@ -1,304 +1,202 @@ - 4.0.0 - cdc-debezium-source - 3.0.0-SNAPSHOT - cdc-debezium-source - CDC Debezium source apps - jar + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> + 4.0.0 - - org.springframework.cloud.stream.app - stream-applications-core - 3.0.0-SNAPSHOT - - + + org.springframework.cloud.stream.app + stream-applications-core + 3.1.0-SNAPSHOT + + - - 1.25.1 - + + 1.25.1 + - - - org.springframework.cloud.fn - cdc-debezium-supplier - ${java-functions.version} - - - slf4j-log4j12 - org.slf4j - - - - - org.springframework.cloud.fn - function-test-support - ${java-functions.version} - test - - - ch.qos.logback - logback-classic - - - - - org.springframework.kafka - spring-kafka - - - org.springframework - spring-jdbc - test - - - net.javacrumbs.json-unit - json-unit - ${json-unit.version} - test - + cdc-debezium-source + cdc-debezium-source + CDC Debezium source apps + jar + + + + org.springframework.cloud.fn + cdc-debezium-supplier + ${java-functions.version} + + + slf4j-log4j12 + org.slf4j + + + + + org.springframework.cloud.fn + function-test-support + ${java-functions.version} + test + + + ch.qos.logback + logback-classic + + + + + org.springframework.cloud.stream.app + stream-applications-test-support + ${stream-apps-core.version} + test + + + org.springframework.kafka + spring-kafka + + + org.springframework + spring-jdbc + test + + + net.javacrumbs.json-unit + json-unit + ${json-unit.version} + test + org.springframework.cloud spring-cloud-stream-binder-kafka test + + org.testcontainers + testcontainers + test + + + + org.junit.jupiter + junit-jupiter-params + test + + + + org.testcontainers + junit-jupiter + test + + + + org.testcontainers + mysql + + + + org.springframework.cloud.fn + cdc-debezium-boot-starter + 1.1.0-SNAPSHOT + test + + - - - - org.springframework.cloud - spring-cloud-dataflow-apps-docs-plugin - - - org.springframework.cloud - spring-cloud-dataflow-apps-generator-plugin - - - cdc-debezium - source - ${project.version} - org.springframework.cloud.fn.supplier.cdc.CdcSupplierConfiguration.class - cdcSupplier - - - - - org.springframework.cloud.fn - cdc-debezium-supplier - - - - - - - - io.fabric8 - docker-maven-plugin - 0.33.0 - - ${skipTests} - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - mysql - debezium/example-mysql:1.0 - - - debezium - mysqluser - mysqlpw - - - 3306:3306 - - - port: 3306 - - - - - - postgres - debezium/example-postgres:1.0 - - - postgres - postgres - - - 5432:5432 - - - PostgreSQL init process complete - - - - - - - properties - docker - override - - %a/sqlserver-example:${project.version} - - ${project.basedir}/src/test/docker/sqlserver - @ - - - - Y - Standard - Password! - true - - - 1433:1433 - - - 1 rows affected - - - - - - - - - start - - generate-test-resources - - build - start - - - - stop - post-integration-test - - stop - - - - stop-pre - clean - - stop - - - - - - - - - - - true - - spring-snapshots - Spring Snapshots - https://repo.spring.io/libs-snapshot-local - - - - false - - spring-milestones - Spring Milestones - https://repo.spring.io/libs-milestone-local - - - - false - - spring-releases - Spring Releases - https://repo.spring.io/release - - - - false - - spring-libs-release - Spring Libs Release - https://repo.spring.io/libs-release - - - - false - - spring-milestone-release - Spring Milestone Release - https://repo.spring.io/libs-milestone - - - - - spring-releases - Spring Releases - https://repo.spring.io/libs-release - - - - true - - spring-snapshots - Spring Snapshots - https://repo.spring.io/libs-snapshot-local - - - - false - - spring-milestones - Spring Milestones - https://repo.spring.io/libs-milestone-local - - + + + + org.springframework.cloud + spring-cloud-dataflow-apps-docs-plugin + + + org.springframework.cloud + spring-cloud-dataflow-apps-generator-plugin + + + cdc-debezium + source + ${project.version} + org.springframework.cloud.fn.supplier.cdc.CdcSupplierConfiguration.class + + cdcSupplier + + + + org.springframework.cloud.fn + cdc-debezium-supplier + + + + + + + + + + + + true + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + + + false + + spring-releases + Spring Releases + https://repo.spring.io/release + + + + false + + spring-libs-release + Spring Libs Release + https://repo.spring.io/libs-release + + + + false + + spring-milestone-release + Spring Milestone Release + https://repo.spring.io/libs-milestone + + + + + spring-releases + Spring Releases + https://repo.spring.io/libs-release + + + + true + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + diff --git a/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcDeleteHandlingIntegrationTest.java b/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcDeleteHandlingIntegrationTest.java index 79f9e3d3..2d3b4cda 100644 --- a/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcDeleteHandlingIntegrationTest.java +++ b/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcDeleteHandlingIntegrationTest.java @@ -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 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); - } - } diff --git a/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcFlatteringIntegrationTest.java b/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcFlatteringIntegrationTest.java index 177d8494..3d97ef97 100644 --- a/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcFlatteringIntegrationTest.java +++ b/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcFlatteringIntegrationTest.java @@ -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 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> messages = receiveAll(outputDestination); assertThat(messages).hasSizeGreaterThanOrEqualTo(52); @@ -169,7 +162,8 @@ public class CdcFlatteringIntegrationTest { final ContextConsumer 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> 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"); } } 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 37035c90..23e19a14 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 @@ -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 { +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 { "--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> messages = receiveAll(outputDestination); @@ -67,17 +74,32 @@ public class CdcSourceDatabasesIntegrationTest { @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> messages = receiveAll(outputDestination); @@ -88,27 +110,40 @@ public class CdcSourceDatabasesIntegrationTest { @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> 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 { assertThat(messages).isNotNull(); assertThat(messages).hasSize(666); } + mongodb.stop(); } } diff --git a/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcTestSupport.java b/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcTestSupport.java new file mode 100644 index 00000000..c879f2e0 --- /dev/null +++ b/applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcTestSupport.java @@ -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"); + } +} diff --git a/applications/source/cdc-debezium-source/src/test/resources/logback-test.xml b/applications/source/cdc-debezium-source/src/test/resources/logback-test.xml new file mode 100644 index 00000000..96fa1fcd --- /dev/null +++ b/applications/source/cdc-debezium-source/src/test/resources/logback-test.xml @@ -0,0 +1,30 @@ + + + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n + + + + + + + + + + \ No newline at end of file diff --git a/applications/source/cdc-debezium-source/src/test/docker/sqlserver/Dockerfile b/applications/source/cdc-debezium-source/src/test/resources/sqlserver/Dockerfile similarity index 86% rename from applications/source/cdc-debezium-source/src/test/docker/sqlserver/Dockerfile rename to applications/source/cdc-debezium-source/src/test/resources/sqlserver/Dockerfile index 75f022b7..071156ba 100644 --- a/applications/source/cdc-debezium-source/src/test/docker/sqlserver/Dockerfile +++ b/applications/source/cdc-debezium-source/src/test/resources/sqlserver/Dockerfile @@ -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 diff --git a/applications/source/cdc-debezium-source/src/test/docker/sqlserver/entrypoint.sh b/applications/source/cdc-debezium-source/src/test/resources/sqlserver/entrypoint.sh similarity index 100% rename from applications/source/cdc-debezium-source/src/test/docker/sqlserver/entrypoint.sh rename to applications/source/cdc-debezium-source/src/test/resources/sqlserver/entrypoint.sh diff --git a/applications/source/cdc-debezium-source/src/test/docker/sqlserver/import-data.sh b/applications/source/cdc-debezium-source/src/test/resources/sqlserver/import-data.sh similarity index 100% rename from applications/source/cdc-debezium-source/src/test/docker/sqlserver/import-data.sh rename to applications/source/cdc-debezium-source/src/test/resources/sqlserver/import-data.sh diff --git a/applications/source/cdc-debezium-source/src/test/docker/sqlserver/inventory.sql b/applications/source/cdc-debezium-source/src/test/resources/sqlserver/inventory.sql similarity index 100% rename from applications/source/cdc-debezium-source/src/test/docker/sqlserver/inventory.sql rename to applications/source/cdc-debezium-source/src/test/resources/sqlserver/inventory.sql diff --git a/applications/stream-applications-core/stream-applications-test-support/pom.xml b/applications/stream-applications-core/stream-applications-test-support/pom.xml index a9148984..ef58dee1 100644 --- a/applications/stream-applications-core/stream-applications-test-support/pom.xml +++ b/applications/stream-applications-core/stream-applications-test-support/pom.xml @@ -28,28 +28,24 @@ org.testcontainers testcontainers - ${test-containers.version} true org.testcontainers junit-jupiter - ${test-containers.version} true org.testcontainers kafka - ${test-containers.version} true org.testcontainers rabbitmq - ${test-containers.version} true diff --git a/functions/common/cdc-debezium-boot-starter/pom.xml b/functions/common/cdc-debezium-boot-starter/pom.xml index 03c27d16..fa58ed95 100644 --- a/functions/common/cdc-debezium-boot-starter/pom.xml +++ b/functions/common/cdc-debezium-boot-starter/pom.xml @@ -45,17 +45,17 @@ ${version.debezium} - - - - - - - - - - - + + + + + + + + + + + io.debezium @@ -97,68 +97,28 @@ 8.0.13 + + org.testcontainers + junit-jupiter + test + + + + org.testcontainers + mysql + + org.springframework spring-jdbc test - + + com.zaxxer + HikariCP + test + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + 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 c489ad8a..ff183b74 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 @@ -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); diff --git a/functions/common/cdc-debezium-boot-starter/src/test/java/org/springframework/cloud/fn/common/cdc/TestCdcApplication.java b/functions/common/cdc-debezium-boot-starter/src/test/java/org/springframework/cloud/fn/common/cdc/TestCdcApplication.java index bf12cd31..79c79a71 100644 --- a/functions/common/cdc-debezium-boot-starter/src/test/java/org/springframework/cloud/fn/common/cdc/TestCdcApplication.java +++ b/functions/common/cdc-debezium-boot-starter/src/test/java/org/springframework/cloud/fn/common/cdc/TestCdcApplication.java @@ -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 { private final Function valueSerializer; + private final Function keySerializer; public Map keyValue = new HashMap<>(); - public List recordList = new ArrayList<>(); + + public List recordList = new CopyOnWriteArrayList<>(); public TestSourceRecordConsumer(Function valueSerializer, Function keySerializer) { @@ -75,5 +78,3 @@ public class TestCdcApplication { } } } - - diff --git a/functions/common/cdc-debezium-boot-starter/src/test/resources/logback-test.xml b/functions/common/cdc-debezium-boot-starter/src/test/resources/logback-test.xml new file mode 100644 index 00000000..96fa1fcd --- /dev/null +++ b/functions/common/cdc-debezium-boot-starter/src/test/resources/logback-test.xml @@ -0,0 +1,30 @@ + + + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n + + + + + + + + + + \ No newline at end of file diff --git a/functions/spring-functions-parent/pom.xml b/functions/spring-functions-parent/pom.xml index fcc7b15a..cb4f7739 100644 --- a/functions/spring-functions-parent/pom.xml +++ b/functions/spring-functions-parent/pom.xml @@ -75,6 +75,18 @@ test + + org.testcontainers + junit-jupiter + test + + + + org.testcontainers + testcontainers + test + + org.awaitility awaitility diff --git a/functions/supplier/cdc-debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/cdc/CdcSupplierConfiguration.java b/functions/supplier/cdc-debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/cdc/CdcSupplierConfiguration.java index c2a6c60a..9055e3fa 100644 --- a/functions/supplier/cdc-debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/cdc/CdcSupplierConfiguration.java +++ b/functions/supplier/cdc-debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/cdc/CdcSupplierConfiguration.java @@ -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> 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()); }; diff --git a/stream-applications-build/pom.xml b/stream-applications-build/pom.xml index acf52544..eaa7a076 100644 --- a/stream-applications-build/pom.xml +++ b/stream-applications-build/pom.xml @@ -74,6 +74,13 @@ pom import + + org.testcontainers + testcontainers-bom + ${test-containers.version} + pom + import +