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

fix regressions. Add DB2 Itests

Browse Source 
Signed-off-by: Christian Tzolov <christian.tzolov@gmail.com>
This commit is contained in:
Christian Tzolov
2022-08-05 16:05:28 +02:00
parent 641e9886e6
commit ae3c0ab6a7
9 changed files with 91 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
applications/source/cdc-debezium-source/README.adoc
View File

@@ -23,7 +23,7 @@ Properties grouped by prefix:
=== cdc
$$config$$:: $$Spring pass-trough wrapper for debezium configuration properties. All properties with a 'cdc.config.' prefix are native Debezium properties. The prefix is removed, converting them into Debezium io.debezium.config.Configuration.$$ *($$Map<String, String>$$, default: `$$<none>$$`)*
$$connector$$:: $$Shortcut for the cdc.config.connector.class property. Either of those can be used as long as they do not contradict with each other.$$ *($$ConnectorType$$, default: `$$<none>$$`, possible values: `mysql`,`postgres`,`mongodb`,`oracle`,`sqlserver`)*
$$connector$$:: $$Shortcut for the cdc.config.connector.class property. Either of those can be used as long as they do not contradict with each other.$$ *($$ConnectorType$$, default: `$$<none>$$`, possible values: `mysql`,`postgres`,`mongodb`,`oracle`,`db2`,`sqlserver`)*
$$name$$:: $$Unique name for this sourceConnector instance.$$ *($$String$$, default: `$$<none>$$`)*
$$schema$$:: $$Include the schema's as part of the outbound message.$$ *($$Boolean$$, default: `$$false$$`)*

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

@@ -212,7 +212,7 @@ public class CdcFlatteningIntegrationTest extends CdcMySqlTestSupport {
assertThat(messages.get(1).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector.inventory.customers");
JsonAssert.assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(1).getHeaders().get("cdc_key")));
if (!StringUtils.hasText(flatteningProps.getAddHeaders()) && flatteningProps.getAddHeaders().contains("op")) {
if (StringUtils.hasText(flatteningProps.getAddHeaders()) && flatteningProps.getAddHeaders().contains("op")) {
assertThat(messages.get(1).getHeaders().get("__op")).isEqualTo("u");
}
@@ -220,7 +220,7 @@ public class CdcFlatteningIntegrationTest extends CdcMySqlTestSupport {
assertThat(toString(messages.get(2).getPayload())).isEqualTo("null");
assertThat(messages.get(1).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector.inventory.customers");
JsonAssert.assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(1).getHeaders().get("cdc_key")));
if (!StringUtils.hasText(flatteningProps.getAddHeaders())
if (StringUtils.hasText(flatteningProps.getAddHeaders())
&& flatteningProps.getAddHeaders().contains("op")) {
assertThat(messages.get(2).getHeaders().get("__op")).isEqualTo("d");
}

2
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-2021 the original author or authors.
* Copyright 2020-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.

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

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2021 the original author or authors.
* Copyright 2020-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -154,6 +154,56 @@ public class CdcSourceDatabasesIntegrationTest {
postgres.stop();
}
// From within the src/test/docker/db2 folder run:
// docker build -t db2-cdc2 .
//
// docker run -itd --name mydb2 --privileged=true -p 50000:50000 -e LICENSE=accept -e DB2INST1_PASSWORD=password -e
// DBNAME=testdb db2-cdc2
//
// docker logs -f mydb2
// docker stop mydb2
// docker exec -it mydb2 /bin/bash
@Test
@Disabled
public void db2() {
// TODO
// GenericContainer db2Container = new GenericContainer(
// new ImageFromDockerfile()
// .withFileFromPath(".", file.toPath()))
// .withPrivilegedMode(true)
// .withExposedPorts(50000)
// .withEnv("LICENSE", "accept")
// .withEnv("DBNAME", "testdb")
// .withEnv("DB2INST1_PASSWORD", "password");
// db2Container.start();
try (ConfigurableApplicationContext context = applicationBuilder
.run("--cdc.connector=db2",
"--cdc.config.database.user=db2inst1",
"--cdc.config.database.password=password",
"--cdc.config.slot.name=debezium",
"--cdc.config.database.dbname=TESTDB",
"--cdc.config.database.hostname=localhost",
// "--cdc.config.table.include.list=inventory.*",
"--cdc.config.database.port=" + "50000")) {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
// Using local region here
List<Message<?>> allMessages = new ArrayList<>();
Awaitility.await().atMost(Duration.ofMinutes(5)).until(() -> {
List<Message<?>> messageChunk = CdcTestUtils.receiveAll(outputDestination);
if (!CollectionUtils.isEmpty(messageChunk)) {
System.out.println("Chunk size: " + messageChunk.size());
allMessages.addAll(messageChunk);
}
return allMessages.size() == 30; // Inventory DB entries
});
}
// db2Container.stop();
}
@Test
@Disabled
public void mongodb() {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2021 the original author or authors.
* Copyright 2020-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.

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

@@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 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>
@@ -19,6 +19,11 @@
</properties>
<dependencies>
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-api</artifactId>
<version>${version.debezium}</version>
</dependency>
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-embedded</artifactId>
@@ -60,6 +65,12 @@
<optional>true</optional>
<version>${version.debezium}</version>
</dependency>
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-db2</artifactId>
<optional>true</optional>
<version>${version.debezium}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>

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

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -64,10 +64,10 @@ public class CdcCommonConfiguration {
Map<String, Object> config = extractNewRecordState.config().defaultValues();
config.put("drop.tombstones", properties.getFlattening().isDropTombstones());
config.put("delete.handling.mode", properties.getFlattening().getDeleteHandlingMode().name());
if (!StringUtils.hasText(properties.getFlattening().getAddHeaders())) {
if (StringUtils.hasText(properties.getFlattening().getAddHeaders())) {
config.put("add.headers", properties.getFlattening().getAddHeaders());
}
if (!StringUtils.hasText(properties.getFlattening().getAddFields())) {
if (StringUtils.hasText(properties.getFlattening().getAddFields())) {
config.put("add.fields", properties.getFlattening().getAddFields());
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.

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

@@ -1,5 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 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>
@@ -78,5 +79,21 @@
</exclusions>
<version>${version.debezium}</version>
</dependency>
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-db2</artifactId>
<exclusions>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
<version>${version.debezium}</version>
</dependency>
<dependency>
<groupId>com.ibm.db2.jcc</groupId>
<artifactId>db2jcc</artifactId>
<version>db2jcc4</version>
</dependency>
</dependencies>
</project>