GH-8632: Add DSL for Debezium module

Fixes #8632

* Debezium DSL initial support
* additional dsl debezium factory
* debezium dsl improvements and tests
* impove debezium docs and streamline dsl testing
* docs clarifications
* fix doc cross-reference
* updgrade debezium to 2.2.1.Final. Clean docs
* fix multiflow config tests
* improve batch tests
* Code and doc formatting
* Make `name` Debezium property as random according to its docs:
```
Unique name for the connector.
Attempting to register again with the same name fails.
This property is required by all Kafka Connect connectors.
```
* Code style clean up

build.gradle

@@ -60,7 +60,7 @@ ext {
     commonsIoVersion = '2.11.0'
     commonsNetVersion = '3.9.0'
     curatorVersion = '5.5.0'
-    debeziumVersion = '2.2.0.Final'
+    debeziumVersion = '2.2.1.Final'
     derbyVersion = '10.16.1.1'
     findbugsVersion = '3.0.1'
     ftpServerVersion = '1.2.0'

spring-integration-debezium/src/main/java/org/springframework/integration/debezium/dsl/Debezium.java

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2023-2023 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.integration.debezium.dsl;
+
+import java.util.Properties;
+
+import io.debezium.engine.ChangeEvent;
+import io.debezium.engine.DebeziumEngine;
+import io.debezium.engine.format.JsonByteArray;
+import io.debezium.engine.format.KeyValueHeaderChangeEventFormat;
+import io.debezium.engine.format.SerializationFormat;
+
+import org.springframework.util.Assert;
+
+/**
+ * Factory class for Debezium DSL components.
+ *
+ * @author Christian Tzolov
+ * @author Artem Bilan
+ *
+ * @since 6.2
+ */
+public final class Debezium {
+
+	/**
+	 * Create an instance of {@link DebeziumMessageProducerSpec} for the provided native debezium {@link Properties} and
+	 * JSON serialization formats.
+	 * @param debeziumConfig {@link Properties} with required debezium engine and connector properties.
+	 * @return the spec.
+	 */
+	public static DebeziumMessageProducerSpec inboundChannelAdapter(Properties debeziumConfig) {
+		return inboundChannelAdapter(debeziumConfig, JsonByteArray.class, JsonByteArray.class);
+	}
+
+	/**
+	 * Create an instance of {@link DebeziumMessageProducerSpec} for the provided native debezium {@link Properties} and
+	 * serialization formats.
+	 * @param debeziumConfig {@link Properties} with required debezium engine and connector properties.
+	 * @param messageFormat  {@link SerializationFormat} format for the {@link ChangeEvent} key and payload.
+	 * @param headerFormat   {@link SerializationFormat} format for the {@link ChangeEvent} headers.
+	 * @return the spec.
+	 */
+	public static DebeziumMessageProducerSpec inboundChannelAdapter(Properties debeziumConfig,
+			Class<? extends SerializationFormat<byte[]>> messageFormat,
+			Class<? extends SerializationFormat<byte[]>> headerFormat) {
+
+		return inboundChannelAdapter(builder(debeziumConfig, messageFormat, headerFormat));
+	}
+
+	/**
+	 * Create an instance of {@link DebeziumMessageProducerSpec} for the provided {@link DebeziumEngine.Builder}.
+	 * @param debeziumEngineBuilder the {@link DebeziumEngine.Builder} to use.
+	 * @return the spec.
+	 */
+	public static DebeziumMessageProducerSpec inboundChannelAdapter(
+			DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder) {
+
+		return new DebeziumMessageProducerSpec(debeziumEngineBuilder);
+	}
+
+	private static DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> builder(Properties debeziumConfig,
+			Class<? extends SerializationFormat<byte[]>> messageFormat,
+			Class<? extends SerializationFormat<byte[]>> headerFormat) {
+
+		Assert.notNull(messageFormat, "'messageFormat' must not be null");
+		Assert.notNull(headerFormat, "'headerFormat' must not be null");
+		Assert.notNull(debeziumConfig, "'debeziumConfig' must not be null");
+		Assert.isTrue(debeziumConfig.containsKey("connector.class"), "The 'connector.class' property must be set");
+
+		return DebeziumEngine
+				.create(KeyValueHeaderChangeEventFormat.of(messageFormat, messageFormat, headerFormat))
+				.using(debeziumConfig);
+	}
+
+	private Debezium() {
+	}
+
+}

spring-integration-debezium/src/main/java/org/springframework/integration/debezium/dsl/DebeziumMessageProducerSpec.java

@@ -0,0 +1,127 @@
+/*
+ * Copyright 2023-2023 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.integration.debezium.dsl;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ThreadFactory;
+
+import io.debezium.engine.ChangeEvent;
+import io.debezium.engine.DebeziumEngine;
+import io.debezium.engine.Header;
+import io.debezium.engine.format.SerializationFormat;
+
+import org.springframework.integration.debezium.inbound.DebeziumMessageProducer;
+import org.springframework.integration.debezium.support.DefaultDebeziumHeaderMapper;
+import org.springframework.integration.dsl.MessageProducerSpec;
+import org.springframework.messaging.support.HeaderMapper;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
+
+/**
+ * A {@link org.springframework.integration.dsl.MessageProducerSpec} for {@link DebeziumMessageProducer}.
+ *
+ * @author Christian Tzolov
+ *
+ * @since 6.2
+ */
+public class DebeziumMessageProducerSpec
+		extends MessageProducerSpec<DebeziumMessageProducerSpec, DebeziumMessageProducer> {
+
+	protected DebeziumMessageProducerSpec(DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder) {
+		super(new DebeziumMessageProducer(debeziumEngineBuilder));
+	}
+
+	/**
+	 * Enable the {@link ChangeEvent} batch mode handling. When enabled the channel adapter will send a {@link List} of
+	 * {@link ChangeEvent}s as a payload in a single downstream {@link org.springframework.messaging.Message}.
+	 * Such a batch payload is not serializable.
+	 * By default, the batch mode is disabled, e.g. every input {@link ChangeEvent} is converted into a
+	 * single downstream {@link org.springframework.messaging.Message}.
+	 * @param enable set to true to enable the batch mode. Disabled by default.
+	 * @return the spec.
+	 */
+	public DebeziumMessageProducerSpec enableBatch(boolean enable) {
+		this.target.setEnableBatch(enable);
+		return this;
+	}
+
+	/**
+	 * Enable support for tombstone (aka delete) messages. On a database row delete, Debezium can send a tombstone
+	 * change event that has the same key as the deleted row and a value of {@link Optional#empty()}. This record is a
+	 * marker for downstream processors. It indicates that log compaction can remove all records that have this key.
+	 * When the tombstone functionality is enabled in the Debezium connector configuration you should enable the empty
+	 * payload as well.
+	 * @param enabled set true to enable the empty payload. Disabled by default.
+	 * @return the spec.
+	 */
+	public DebeziumMessageProducerSpec enableEmptyPayload(boolean enabled) {
+		this.target.setEnableEmptyPayload(enabled);
+		return this;
+	}
+
+	/**
+	 * Set a {@link ThreadFactory} for the Debezium executor. Defaults to the {@link CustomizableThreadFactory} with a
+	 * {@code debezium:inbound-channel-adapter-thread-} prefix.
+	 * @param threadFactory the {@link ThreadFactory} instance to use.
+	 * @return the spec.
+	 */
+	public DebeziumMessageProducerSpec threadFactory(ThreadFactory threadFactory) {
+		this.target.setThreadFactory(threadFactory);
+		return this;
+	}
+
+	/**
+	 * Set the outbound message content type. Must be aligned with the {@link SerializationFormat} configuration used by
+	 * the provided {@link DebeziumEngine}.
+	 * @param contentType payload content type.
+	 * @return the spec.
+	 */
+	public DebeziumMessageProducerSpec contentType(String contentType) {
+		this.target.setContentType(contentType);
+		return this;
+	}
+
+	/**
+	 * Comma-separated list of names of {@link ChangeEvent} headers to be mapped into outbound Message headers.
+	 * Debezium's NewRecordStateExtraction 'add.headers' property configures the metadata to be used as
+	 * {@link ChangeEvent} headers.
+	 * <p>
+	 * You should prefix the names passed to the 'headerNames' with the prefix configured by the Debezium
+	 * 'add.headers.prefix' property. Later defaults to '__'. For example for 'add.headers=op,name' and
+	 * 'add.headers.prefix=__' you should use header hames like: '__op', '__name'.
+	 * @param headerNames The values in this list can be a simple patterns to be matched against the header names.
+	 * @return the spec.
+	 */
+	public DebeziumMessageProducerSpec headerNames(String... headerNames) {
+		DefaultDebeziumHeaderMapper headerMapper = new DefaultDebeziumHeaderMapper();
+		headerMapper.setHeaderNamesToMap(headerNames);
+
+		return headerMapper(headerMapper);
+	}
+
+	/**
+	 * Set a {@link HeaderMapper} to convert the {@link ChangeEvent} headers
+	 * into {@link org.springframework.messaging.Message} headers.
+	 * @param headerMapper {@link HeaderMapper} implementation to use. Defaults to {@link DefaultDebeziumHeaderMapper}.
+	 * @return the spec.
+	 */
+	public DebeziumMessageProducerSpec headerMapper(HeaderMapper<List<Header<Object>>> headerMapper) {
+		this.target.setHeaderMapper(headerMapper);
+		return this;
+	}
+
+}

spring-integration-debezium/src/main/java/org/springframework/integration/debezium/dsl/package-info.java

@@ -0,0 +1,6 @@
+/**
+ * Provides classes for supporting Debezium component via Java DSL.
+ */
+@org.springframework.lang.NonNullApi
+@org.springframework.lang.NonNullFields
+package org.springframework.integration.debezium.dsl;

spring-integration-debezium/src/main/java/org/springframework/integration/debezium/support/DebeziumHeaders.java

@@ -20,6 +20,7 @@
  * Pre-defined header names to be used when retrieving Debezium Change Event headers.
  *
  * @author Christian Tzolov
+ *
  * @since 6.2
  */
 public abstract class DebeziumHeaders {

spring-integration-debezium/src/main/java/org/springframework/integration/debezium/support/DefaultDebeziumHeaderMapper.java

@@ -39,7 +39,7 @@
  */
 public class DefaultDebeziumHeaderMapper implements HeaderMapper<List<Header<Object>>> {
 
-	private String[] headerNamesToMap = new String[0];
+	private String[] headerNamesToMap = {"*"};
 
 	/**
 	 * Comma-separated list of names of Debezium's Change Event headers to be mapped to the outbound Message headers.

spring-integration-debezium/src/test/java/org/springframework/integration/debezium/DebeziumMySqlTestContainer.java

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2023-2023 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.integration.debezium;
+
+import java.util.Properties;
+import java.util.Random;
+
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+/**
+ * @author Christian Tzolov
+ * @author Artem Bilan
+ *
+ * @since 6.2
+ */
+@Testcontainers(disabledWithoutDocker = true)
+public interface DebeziumMySqlTestContainer {
+
+	int EXPECTED_DB_TX_COUNT = 52;
+
+	@Container
+	GenericContainer<?> DEBEZIUM_MYSQL = new GenericContainer<>("debezium/example-mysql:2.2.0.Final")
+			.withExposedPorts(3306)
+			.withEnv("MYSQL_ROOT_PASSWORD", "debezium")
+			.withEnv("MYSQL_USER", "mysqluser")
+			.withEnv("MYSQL_PASSWORD", "mysqlpw")
+			.waitingFor(new LogMessageWaitStrategy().withRegEx(".*port: 3306  MySQL Community Server - GPL.*."));
+
+	static int mysqlPort() {
+		return DEBEZIUM_MYSQL.getMappedPort(3306);
+	}
+
+	static Properties connectorConfig(int port) {
+		Random random = new Random();
+
+		Properties config = new Properties();
+
+		config.put("transforms", "unwrap");
+		config.put("transforms.unwrap.type", "io.debezium.transforms.ExtractNewRecordState");
+		config.put("transforms.unwrap.drop.tombstones", "false");
+		config.put("transforms.unwrap.delete.handling.mode", "rewrite");
+		config.put("transforms.unwrap.add.fields", "name,db,op,table");
+		config.put("transforms.unwrap.add.headers", "name,db,op,table");
+
+		config.put("schema.history.internal", "io.debezium.relational.history.MemorySchemaHistory");
+		config.put("offset.storage", "org.apache.kafka.connect.storage.MemoryOffsetBackingStore");
+
+		config.put("name", "my-connector-" + random.nextInt(10));
+
+		// Topic prefix for the database server or cluster.
+		config.put("topic.prefix", "my-topic-" + random.nextInt(10));
+		// Unique ID of the connector.
+		config.put("database.server.id", "8574" + random.nextInt(10));
+
+		config.put("key.converter.schemas.enable", "false");
+		config.put("value.converter.schemas.enable", "false");
+
+		config.put("connector.class", "io.debezium.connector.mysql.MySqlConnector");
+		config.put("database.user", "debezium");
+		config.put("database.password", "dbz");
+		config.put("database.hostname", "localhost");
+		config.put("database.port", String.valueOf(port));
+
+		return config;
+	}
+
+}