GH-3779: Add Debezium Channel Adapter

Fixes https://github.com/spring-projects/spring-integration/issues/3779

initial debezium doc
address some reviews
resolve some classpath conflicts
hacking failed test
fixing tests and dependecies
address review comments
improve test coverage
fix test checkstyle
remove kafak references. hit support for batch
improve java doc
Initial batch support

 Convert the list of Change events into list of Messages.
 Use the same rules for buidling messages as the non-batch mode.

Refine batch implementation and tests
harden the testcontainers start/stop lifecycle
simplify batch mode
adjust test log config
clean gradle config
Add  `HeaderMapper` filter configuration. Fix JavaDocs
Use `CustomizableThreadFactory` for Exec Service. IT header tests
more debeizum documentation
Remove external Executor support in favor of configurable ThreadFactory
minor `Threadfactory` naming fix
fix support package structure
* Clean up code style and language typos

spring-integration-debezium/src/main/java/org/springframework/integration/debezium/inbound/DebeziumMessageProducer.java

@@ -0,0 +1,292 @@
+/*
+ * 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.inbound;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+
+import io.debezium.engine.ChangeEvent;
+import io.debezium.engine.DebeziumEngine;
+import io.debezium.engine.DebeziumEngine.Builder;
+import io.debezium.engine.DebeziumEngine.ChangeConsumer;
+import io.debezium.engine.DebeziumEngine.RecordCommitter;
+import io.debezium.engine.Header;
+import io.debezium.engine.format.SerializationFormat;
+
+import org.springframework.integration.debezium.support.DebeziumHeaders;
+import org.springframework.integration.debezium.support.DefaultDebeziumHeaderMapper;
+import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.HeaderMapper;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
+import org.springframework.util.Assert;
+
+/**
+ * Debezium Change Event Channel Adapter.
+ *
+ * @author Christian Tzolov
+ * @author Artem Bilan
+ *
+ * @since 6.2
+ */
+public class DebeziumMessageProducer extends MessageProducerSupport {
+
+	private final DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder;
+
+	private DebeziumEngine<ChangeEvent<byte[], byte[]>> debeziumEngine;
+
+	/**
+	 * Debezium Engine is designed to be submitted to an {@link Executor}
+	 * or {@link ExecutorService} for execution by a single thread.
+	 * By default, a single-threaded ExecutorService instance is provided configured with a
+	 * {@link CustomizableThreadFactory} and a `debezium-` thread prefix.
+	 */
+	private ExecutorService executorService;
+
+	private String contentType = "application/json";
+
+	private HeaderMapper<List<Header<Object>>> headerMapper = new DefaultDebeziumHeaderMapper();
+
+	private boolean enableEmptyPayload = false;
+
+	private boolean enableBatch = false;
+
+	private ThreadFactory threadFactory;
+
+	private volatile CountDownLatch latch = new CountDownLatch(0);
+
+	/**
+	 * Create new Debezium message producer inbound channel adapter.
+	 * @param debeziumBuilder - pre-configured Debezium Engine Builder instance.
+	 */
+	public DebeziumMessageProducer(Builder<ChangeEvent<byte[], byte[]>> debeziumBuilder) {
+		Assert.notNull(debeziumBuilder, "'debeziumBuilder' must not be null");
+		this.debeziumEngineBuilder = debeziumBuilder;
+	}
+
+	/**
+	 * 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 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 Message}.
+	 * @param enable set to true to enable the batch mode. Disabled by default.
+	 */
+	public void setEnableBatch(boolean enable) {
+		this.enableBatch = enable;
+	}
+
+	/**
+	 * 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.
+	 */
+	public void setEnableEmptyPayload(boolean enabled) {
+		this.enableEmptyPayload = enabled;
+	}
+
+	/**
+	 * 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.
+	 */
+	public void setThreadFactory(ThreadFactory threadFactory) {
+		Assert.notNull(threadFactory, "'threadFactory' must not be null");
+		this.threadFactory = threadFactory;
+	}
+
+	/**
+	 * Set the outbound message content type.
+	 * Must be aligned with the {@link SerializationFormat} configuration used by
+	 * the provided {@link DebeziumEngine}.
+	 */
+	public void setContentType(String contentType) {
+		Assert.hasText(contentType, "'contentType' must not be empty");
+		this.contentType = contentType;
+	}
+
+	/**
+	 * Set a {@link HeaderMapper} to convert the {@link ChangeEvent}
+	 * headers into {@link Message} headers.
+	 * @param headerMapper {@link HeaderMapper} implementation to use.
+	 * Defaults to {@link DefaultDebeziumHeaderMapper}.
+	 */
+	public void setHeaderMapper(HeaderMapper<List<Header<Object>>> headerMapper) {
+		Assert.notNull(headerMapper, "'headerMapper' must not be null.");
+		this.headerMapper = headerMapper;
+	}
+
+	@Override
+	public String getComponentType() {
+		return "debezium:inbound-channel-adapter";
+	}
+
+	@Override
+	protected void onInit() {
+		super.onInit();
+
+		if (this.threadFactory == null) {
+			this.threadFactory = new CustomizableThreadFactory(getComponentName() + "-thread-");
+		}
+
+		this.executorService = Executors.newSingleThreadExecutor(this.threadFactory);
+
+		if (!this.enableBatch) {
+			this.debeziumEngineBuilder.notifying(new StreamChangeEventConsumer<>());
+		}
+		else {
+			this.debeziumEngineBuilder.notifying(new BatchChangeEventConsumer<>());
+		}
+
+		this.debeziumEngine = this.debeziumEngineBuilder.build();
+	}
+
+	@Override
+	protected void doStart() {
+		if (this.latch.getCount() > 0) {
+			return;
+		}
+		this.latch = new CountDownLatch(1);
+		this.executorService.execute(() -> {
+			try {
+				// Runs the debezium connector and deliver database changes to the registered consumer. This method
+				// blocks until the connector is stopped.
+				// If this instance is already running, then the run immediately returns.
+				// When run the connector and starts polling the configured connector for change events.
+				// All messages are delivered in batches to the consumer registered with this debezium engine.
+				// The batch size, polling frequency, and other parameters are controlled via connector's configuration
+				// settings. This continues until this connector is stopped.
+				// This method can be called repeatedly as needed.
+				this.debeziumEngine.run();
+			}
+			finally {
+				this.latch.countDown();
+			}
+		});
+	}
+
+	@Override
+	protected void doStop() {
+		try {
+			this.debeziumEngine.close();
+		}
+		catch (IOException e) {
+			logger.warn(e, "Debezium failed to close!");
+		}
+		try {
+			if (!this.latch.await(5, TimeUnit.SECONDS)) {
+				throw new IllegalStateException("Failed to stop " + this);
+			}
+		}
+		catch (InterruptedException ignored) {
+		}
+	}
+
+	@Override
+	public void destroy() {
+		super.destroy();
+
+		this.executorService.shutdown();
+		try {
+			this.executorService.awaitTermination(5, TimeUnit.SECONDS);
+		}
+		catch (InterruptedException e) {
+			throw new IllegalStateException("Debezium failed to close!", e);
+		}
+	}
+
+	@Nullable
+	private <T> Message<?> toMessage(ChangeEvent<T, T> changeEvent) {
+		Object key = changeEvent.key();
+		Object payload = changeEvent.value();
+		String destination = changeEvent.destination();
+
+		// 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 (debezium_key).
+		// Note: Event for none flattened responses, when the debezium.properties.tombstones.on.delete=true
+		// (default), tombstones are generate by Debezium and handled by the code below.
+		if (payload == null && DebeziumMessageProducer.this.enableEmptyPayload) {
+			payload = Optional.empty();
+		}
+
+		// If payload is still null ignore the message.
+		if (payload == null) {
+			logger.info(() -> "Dropped null payload message for Change Event key: " + key);
+			return null;
+		}
+
+		return getMessageBuilderFactory()
+				.withPayload(payload)
+				.setHeader(DebeziumHeaders.KEY, key)
+				.setHeader(DebeziumHeaders.DESTINATION, destination)
+				.setHeader(MessageHeaders.CONTENT_TYPE, this.contentType)
+				// Use the provided header mapper to convert Debezium headers into message headers.
+				.copyHeaders(this.headerMapper.toHeaders(changeEvent.headers()))
+				.build();
+	}
+
+	final class StreamChangeEventConsumer<T> implements Consumer<ChangeEvent<T, T>> {
+
+		@Override
+		public void accept(ChangeEvent<T, T> changeEvent) {
+			Message<?> message = toMessage(changeEvent);
+			if (message != null) {
+				sendMessage(message);
+			}
+		}
+
+	}
+
+	final class BatchChangeEventConsumer<T> implements ChangeConsumer<ChangeEvent<T, T>> {
+
+		@Override
+		public void handleBatch(List<ChangeEvent<T, T>> changeEvents, RecordCommitter<ChangeEvent<T, T>> committer)
+				throws InterruptedException {
+
+			Message<List<ChangeEvent<T, T>>> message =
+					getMessageBuilderFactory()
+							.withPayload(changeEvents)
+							.build();
+
+			sendMessage(message);
+
+			for (ChangeEvent<T, T> event : changeEvents) {
+				committer.markProcessed(event);
+			}
+			committer.markBatchFinished();
+		}
+
+	}
+
+}

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

@@ -0,0 +1,5 @@
+/**
+ * Provides classes for the Debezium inbound channel adapters.
+ */
+@org.springframework.lang.NonNullApi
+package org.springframework.integration.debezium.inbound;

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

@@ -0,0 +1,39 @@
+/*
+ * 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.support;
+
+/**
+ * Pre-defined header names to be used when retrieving Debezium Change Event headers.
+ *
+ * @author Christian Tzolov
+ * @since 6.2
+ */
+public abstract class DebeziumHeaders {
+
+	public static final String PREFIX = "debezium_";
+
+	/**
+	 * Debezium's header key.
+	 */
+	public static final String KEY = PREFIX + "key";
+
+	/**
+	 * Debezium's event destination.
+	 */
+	public static final String DESTINATION = PREFIX + "destination";
+
+}

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

@@ -0,0 +1,95 @@
+/*
+ * 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.support;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import io.debezium.engine.Header;
+
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.HeaderMapper;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.PatternMatchUtils;
+
+/**
+ * Specifies how to convert Debezium {@link io.debezium.engine.ChangeEvent#headers()} into a {@link MessageHeaders}.
+ *
+ * @author Christian Tzolov
+ * @author Artem Bilan
+ *
+ * @since 6.2
+ */
+public class DefaultDebeziumHeaderMapper implements HeaderMapper<List<Header<Object>>> {
+
+	private String[] headerNamesToMap = new String[0];
+
+	/**
+	 * Comma-separated list of names of Debezium's Change Event headers to be mapped to the outbound Message headers.
+	 * The Debezium New Record State Extraction 'add.headers' property is used to configure the metadata to be set in
+	 * the produced ChangeEvent headers. Note that you must prefix the 'headerNames' used the 'setHeaderNamesToMap'
+	 * with the prefix configured by the 'add.headers.prefix' debezium property. Later defaults to '__'. For example for
+	 * 'add.headers=op,name' and 'add.headers.prefix=__' you should use headerNames == "__op", "__name".
+	 * @param headerNames The values in this list can be a simple patterns to be matched against the header names.
+	 * @see <a href="https://debezium.io/documentation/reference/transformations/event-flattening.html#extract-new-record-state-add-headers">add.headers</a>
+	 * @see <a href="https://debezium.io/documentation/reference/transformations/event-flattening.html#extract-new-record-state-add-headers-prefix">add.headers.prefix</a>
+	 */
+	public void setHeaderNamesToMap(String... headerNames) {
+		Assert.notNull(headerNames, "'HeaderNames' must not be null.");
+		Assert.noNullElements(headerNames, "'HeaderNames' must not contains null elements.");
+		String[] copy = Arrays.copyOf(headerNames, headerNames.length);
+		Arrays.sort(copy);
+		this.headerNamesToMap = copy;
+	}
+
+	@Override
+	public MessageHeaders toHeaders(List<Header<Object>> debeziumHeaders) {
+		Map<String, Object> messageHeaders = null;
+		if (!CollectionUtils.isEmpty(debeziumHeaders)) {
+			messageHeaders = new HashMap<>();
+			for (Header<Object> header : debeziumHeaders) {
+				String headerName = header.getKey();
+				if (shouldMapHeader(headerName, this.headerNamesToMap)) {
+					Object headerValue = header.getValue();
+					messageHeaders.put(headerName, headerValue);
+				}
+			}
+		}
+		return new MessageHeaders(messageHeaders);
+	}
+
+	@Override
+	public void fromHeaders(MessageHeaders headers, List<Header<Object>> target) {
+		throw new UnsupportedOperationException("The 'fromHeaders' is not supported!");
+	}
+
+	private static boolean shouldMapHeader(String headerName, String[] patterns) {
+		if (patterns.length > 0) {
+			for (String pattern : patterns) {
+				if (PatternMatchUtils.simpleMatch(pattern, headerName)) {
+					return true;
+				}
+			}
+		}
+
+		return false;
+	}
+
+}

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

@@ -0,0 +1,5 @@
+/**
+ * Provides supporting classes for the Debezium channel adapters.
+ */
+@org.springframework.lang.NonNullApi
+package org.springframework.integration.debezium.support;

spring-integration-debezium/src/test/java/org/springframework/integration/debezium/inbound/DebeziumMessageProducerTests.java

@@ -0,0 +1,99 @@
+/*
+ * 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.inbound;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+
+import io.debezium.engine.ChangeEvent;
+import io.debezium.engine.DebeziumEngine;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.integration.channel.QueueChannel;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.then;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.reset;
+
+/**
+ * @author Christian Tzolov
+ * @since 6.2
+ */
+public class DebeziumMessageProducerTests {
+
+	DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> debeziumBuilderMock;
+
+	DebeziumEngine<ChangeEvent<byte[], byte[]>> debeziumEngineMock;
+
+	DebeziumMessageProducer debeziumMessageProducer;
+
+	@BeforeEach
+	@SuppressWarnings("unchecked")
+	public void beforeEach() {
+		debeziumBuilderMock = mock(DebeziumEngine.Builder.class);
+		debeziumEngineMock = mock(DebeziumEngine.class);
+		given(debeziumBuilderMock.notifying(any(Consumer.class))).willReturn(debeziumBuilderMock);
+		given(debeziumBuilderMock.build()).willReturn(debeziumEngineMock);
+
+		debeziumMessageProducer = new DebeziumMessageProducer(debeziumBuilderMock);
+		debeziumMessageProducer.setOutputChannel(new QueueChannel());
+		debeziumMessageProducer.setBeanFactory(mock(BeanFactory.class));
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testDebeziumMessageProducerLifecycle() throws IOException {
+
+		debeziumMessageProducer.afterPropertiesSet(); // INIT
+
+		then(debeziumBuilderMock).should().build();
+		assertThat(debeziumMessageProducer.isActive()).isEqualTo(false);
+		assertThat(debeziumMessageProducer.isRunning()).isEqualTo(false);
+
+		debeziumMessageProducer.start(); // START
+
+		await().atMost(5, TimeUnit.SECONDS).until(() -> debeziumMessageProducer.isRunning());
+		assertThat(debeziumMessageProducer.isActive()).isEqualTo(true);
+		then(debeziumEngineMock).should().run();
+
+		debeziumMessageProducer.stop(); // STOP
+
+		assertThat(debeziumMessageProducer.isActive()).isEqualTo(false);
+		assertThat(debeziumMessageProducer.isRunning()).isEqualTo(false);
+		then(debeziumEngineMock).should().close();
+
+		reset(debeziumEngineMock);
+
+		debeziumMessageProducer.start(); // START
+
+		await().atMost(5, TimeUnit.SECONDS).until(() -> debeziumMessageProducer.isRunning());
+		assertThat(debeziumMessageProducer.isActive()).isEqualTo(true);
+		then(debeziumEngineMock).should().run();
+
+		debeziumMessageProducer.destroy(); // DESTROY
+
+		then(debeziumEngineMock).should().close();
+	}
+
+}

spring-integration-debezium/src/test/java/org/springframework/integration/debezium/it/DebeziumBatchTests.java

@@ -0,0 +1,103 @@
+/*
+ * 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.it;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import io.debezium.engine.ChangeEvent;
+import io.debezium.engine.DebeziumEngine;
+import io.debezium.engine.Header;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.core.MessageProducer;
+import org.springframework.integration.debezium.inbound.DebeziumMessageProducer;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Christian Tzolov
+ * @author Artem Bilan
+ *
+ * @since 6.2
+ */
+@SpringJUnitConfig
+@DirtiesContext
+public class DebeziumBatchTests implements DebeziumMySqlTestContainer {
+
+	@Autowired
+	@Qualifier("queueChannel")
+	private QueueChannel queueChannel;
+
+	@Test
+	@SuppressWarnings("unchecked")
+	void batchMode() {
+		Message<?> message = this.queueChannel.receive(20_000);
+		assertThat(message).isNotNull();
+		List<ChangeEvent<Object, Object>> payload = (List<ChangeEvent<Object, Object>>) message.getPayload();
+		assertThat(payload).hasSize(52);
+
+		for (int i = 0; i < 52; i++) {
+			ChangeEvent<Object, Object> changeEvent = payload.get(i);
+
+			List<String> headerKeys =
+					changeEvent.headers()
+							.stream()
+							.map(Header::getKey)
+							.collect(Collectors.toList());
+
+			if (i < 16) {
+				assertThat(changeEvent.destination()).startsWith("my-topic");
+				assertThat(headerKeys).isEmpty();
+			}
+			else {
+				assertThat(changeEvent.destination()).startsWith("my-topic.inventory");
+				assertThat(headerKeys).hasSize(4).contains("__name", "__db", "__op", "__table");
+			}
+		}
+	}
+
+	@Configuration
+	@EnableIntegration
+	@Import(DebeziumTestConfiguration.class)
+	public static class BatchTestConfiguration {
+
+		@Bean
+		public MessageProducer debeziumMessageProducer(
+				@Qualifier("debeziumInputChannel") MessageChannel debeziumInputChannel,
+				DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder) {
+
+			DebeziumMessageProducer debeziumMessageProducer = new DebeziumMessageProducer(debeziumEngineBuilder);
+			debeziumMessageProducer.setEnableBatch(true);
+			debeziumMessageProducer.setOutputChannel(debeziumInputChannel);
+			return debeziumMessageProducer;
+		}
+
+	}
+
+}

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

@@ -0,0 +1,45 @@
+/*
+ * 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.it;
+
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+/**
+ * @author Christian Tzolov
+ * @author Artem Bilan
+ *
+ * @since 6.2
+ */
+@Testcontainers(disabledWithoutDocker = true)
+interface DebeziumMySqlTestContainer {
+
+	@SuppressWarnings("resource")
+	@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");
+
+	static int mysqlPort() {
+		return DEBEZIUM_MYSQL.getMappedPort(3306);
+	}
+
+}

spring-integration-debezium/src/test/java/org/springframework/integration/debezium/it/DebeziumStreamTests.java

@@ -0,0 +1,93 @@
+/*
+ * 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.it;
+
+import io.debezium.engine.ChangeEvent;
+import io.debezium.engine.DebeziumEngine;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.core.MessageProducer;
+import org.springframework.integration.debezium.inbound.DebeziumMessageProducer;
+import org.springframework.integration.debezium.support.DefaultDebeziumHeaderMapper;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Christian Tzolov
+ * @author Artem Bilan
+ *
+ * @since 6.2
+ */
+@SpringJUnitConfig
+@DirtiesContext
+public class DebeziumStreamTests implements DebeziumMySqlTestContainer {
+
+	@Autowired
+	@Qualifier("queueChannel")
+	private QueueChannel queueChannel;
+
+	@Test
+	void streamMode() {
+		boolean foundDebeziumHeaders = false;
+		for (int i = 0; i < 52; i++) {
+			Message<?> message = this.queueChannel.receive(10_000);
+			assertThat(message).isNotNull();
+
+			if (message.getHeaders().size() > 5) {
+				assertThat(message.getHeaders()).containsKeys("__name", "__db", "__op", "__table");
+				foundDebeziumHeaders = true;
+			}
+		}
+		assertThat(foundDebeziumHeaders).isTrue();
+	}
+
+	@Configuration
+	@EnableIntegration
+	@Import(DebeziumTestConfiguration.class)
+	public static class StreamTestConfiguration {
+
+		@Bean
+		public MessageProducer debeziumMessageProducer(
+				@Qualifier("debeziumInputChannel") MessageChannel debeziumInputChannel,
+				DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder) {
+
+			DebeziumMessageProducer debeziumMessageProducer = new DebeziumMessageProducer(debeziumEngineBuilder);
+
+			// This corresponds to the 'transforms.unwrap.add.headers=name,db,op,table' debezium configuration in
+			// the DebeziumTestConfiguration#debeziumEngineBuilder!
+			DefaultDebeziumHeaderMapper debeziumHeaderMapper = new DefaultDebeziumHeaderMapper();
+			debeziumHeaderMapper.setHeaderNamesToMap("__name", "__db", "__op", "__table");
+			debeziumMessageProducer.setHeaderMapper(debeziumHeaderMapper);
+
+			debeziumMessageProducer.setOutputChannel(debeziumInputChannel);
+			return debeziumMessageProducer;
+		}
+
+	}
+
+}

spring-integration-debezium/src/test/java/org/springframework/integration/debezium/it/DebeziumTestConfiguration.java

@@ -0,0 +1,89 @@
+/*
+ * 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.it;
+
+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 org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.annotation.BridgeFrom;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * @author Christian Tzolov
+ * @author Artem Bilan
+ *
+ * @since 6.2
+ */
+@Configuration
+@EnableIntegration
+public class DebeziumTestConfiguration {
+
+	@Bean
+	public DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder() {
+		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("topic.prefix", "my-topic");
+		config.put("name", "my-connector");
+		config.put("database.server.id", "85744");
+		config.put("database.server.name", "my-app-connector");
+
+		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(DebeziumMySqlTestContainer.mysqlPort()));
+
+		KeyValueHeaderChangeEventFormat<JsonByteArray, JsonByteArray, JsonByteArray> format =
+				KeyValueHeaderChangeEventFormat.of(
+						io.debezium.engine.format.JsonByteArray.class,
+						io.debezium.engine.format.JsonByteArray.class,
+						io.debezium.engine.format.JsonByteArray.class);
+
+		return DebeziumEngine.create(format).using(config);
+	}
+
+	@Bean
+	public MessageChannel debeziumInputChannel() {
+		return new DirectChannel();
+	}
+
+	@Bean
+	@BridgeFrom("debeziumInputChannel")
+	public MessageChannel queueChannel() {
+		return new QueueChannel();
+	}
+
+}

spring-integration-debezium/src/test/java/org/springframework/integration/debezium/support/DefaultDebeziumHeaderMapperTests.java

@@ -0,0 +1,85 @@
+/*
+ * 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.support;
+
+import java.util.List;
+
+import io.debezium.engine.Header;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * @author Christian Tzolov
+ * @author Artem Bilan
+ *
+ * @since 6.2
+ */
+public class DefaultDebeziumHeaderMapperTests {
+
+	final DefaultDebeziumHeaderMapper mapper = new DefaultDebeziumHeaderMapper();
+
+	final List<Header<Object>> debeziumHeaders = List.of(
+			new TestHeader<>("NonStandard1", "NonStandard1_Value"),
+			new TestHeader<>("NonStandard2", "NonStandard2_Value"));
+
+	@Test
+	public void fromHeaderNotSupported() {
+		assertThatThrownBy(() -> mapper.fromHeaders(null, null))
+				.isInstanceOf(UnsupportedOperationException.class)
+				.hasMessage("The 'fromHeaders' is not supported!");
+	}
+
+	@Test
+	public void defaultHeaders() {
+		assertThat(mapper.toHeaders(debeziumHeaders)).hasSize(2)
+				.containsKeys("id", "timestamp");
+	}
+
+	@Test
+	public void exactCustomHeaderName() {
+		mapper.setHeaderNamesToMap("NonStandard1");
+
+		assertThat(mapper.toHeaders(debeziumHeaders)).hasSize(3)
+				.containsKeys("NonStandard1", "id", "timestamp");
+	}
+
+	@Test
+	public void headerNamePattern() {
+		mapper.setHeaderNamesToMap("NonStandard*");
+
+		assertThat(mapper.toHeaders(debeziumHeaders)).hasSize(4)
+				.containsKeys("NonStandard1", "NonStandard2", "id", "timestamp");
+
+	}
+
+	public record TestHeader<T>(String key, T value) implements Header<T> {
+
+		@Override
+		public String getKey() {
+			return key;
+		}
+
+		@Override
+		public T getValue() {
+			return value;
+		}
+
+	}
+
+}

spring-integration-debezium/src/test/resources/log4j2-test.xml

@@ -0,0 +1,16 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="WARN">
+	<Appenders>
+		<Console name="Console" target="SYSTEM_OUT">
+			<PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n" />
+		</Console>
+	</Appenders>
+	<Loggers>
+		<Logger name="org.springframework.integration" level="warn"/>
+		<Logger name="org.springframework.integration.debezium" level="warn"/>
+
+		<Root level="error">
+			<AppenderRef ref="Console" />
+		</Root>
+	</Loggers>
+</Configuration>