diff --git a/build.gradle b/build.gradle index 301b31b6b8..5302f0d9e6 100644 --- a/build.gradle +++ b/build.gradle @@ -60,6 +60,7 @@ ext { commonsIoVersion = '2.11.0' commonsNetVersion = '3.9.0' curatorVersion = '5.5.0' + debeziumVersion = '2.2.0.Final' derbyVersion = '10.16.1.1' findbugsVersion = '3.0.1' ftpServerVersion = '1.2.0' @@ -576,6 +577,21 @@ project('spring-integration-core') { } } +project('spring-integration-debezium') { + description = 'Spring Integration Debezium Support' + dependencies { + api project(':spring-integration-core') + api("io.debezium:debezium-embedded:$debeziumVersion") { + exclude group: 'org.glassfish.jersey.containers', module: 'jersey-container-servlet' + exclude group: 'org.glassfish.jersey.inject', module: 'jersey-hk2' + exclude group: 'jakarta.xml.bind', module: 'jakarta.xml.bind-api' + exclude group: 'jakarta.activation', module: 'jakarta.activation-api' + } + + testImplementation "io.debezium:debezium-connector-mysql:$debeziumVersion" + } +} + project('spring-integration-event') { description = 'Spring Integration ApplicationEvent Support' dependencies { diff --git a/spring-integration-debezium/src/main/java/org/springframework/integration/debezium/inbound/DebeziumMessageProducer.java b/spring-integration-debezium/src/main/java/org/springframework/integration/debezium/inbound/DebeziumMessageProducer.java new file mode 100644 index 0000000000..1894bc7e48 --- /dev/null +++ b/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> debeziumEngineBuilder; + + private DebeziumEngine> 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>> 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> 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>> 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 Message toMessage(ChangeEvent 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 implements Consumer> { + + @Override + public void accept(ChangeEvent changeEvent) { + Message message = toMessage(changeEvent); + if (message != null) { + sendMessage(message); + } + } + + } + + final class BatchChangeEventConsumer implements ChangeConsumer> { + + @Override + public void handleBatch(List> changeEvents, RecordCommitter> committer) + throws InterruptedException { + + Message>> message = + getMessageBuilderFactory() + .withPayload(changeEvents) + .build(); + + sendMessage(message); + + for (ChangeEvent event : changeEvents) { + committer.markProcessed(event); + } + committer.markBatchFinished(); + } + + } + +} diff --git a/spring-integration-debezium/src/main/java/org/springframework/integration/debezium/inbound/package-info.java b/spring-integration-debezium/src/main/java/org/springframework/integration/debezium/inbound/package-info.java new file mode 100644 index 0000000000..69eb77dc7a --- /dev/null +++ b/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; diff --git a/spring-integration-debezium/src/main/java/org/springframework/integration/debezium/support/DebeziumHeaders.java b/spring-integration-debezium/src/main/java/org/springframework/integration/debezium/support/DebeziumHeaders.java new file mode 100644 index 0000000000..dfc1a7f784 --- /dev/null +++ b/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"; + +} diff --git a/spring-integration-debezium/src/main/java/org/springframework/integration/debezium/support/DefaultDebeziumHeaderMapper.java b/spring-integration-debezium/src/main/java/org/springframework/integration/debezium/support/DefaultDebeziumHeaderMapper.java new file mode 100644 index 0000000000..d45cc47238 --- /dev/null +++ b/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>> { + + 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 add.headers + * @see add.headers.prefix + */ + 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> debeziumHeaders) { + Map messageHeaders = null; + if (!CollectionUtils.isEmpty(debeziumHeaders)) { + messageHeaders = new HashMap<>(); + for (Header 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> 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; + } + +} diff --git a/spring-integration-debezium/src/main/java/org/springframework/integration/debezium/support/package-info.java b/spring-integration-debezium/src/main/java/org/springframework/integration/debezium/support/package-info.java new file mode 100644 index 0000000000..209a295852 --- /dev/null +++ b/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; diff --git a/spring-integration-debezium/src/test/java/org/springframework/integration/debezium/inbound/DebeziumMessageProducerTests.java b/spring-integration-debezium/src/test/java/org/springframework/integration/debezium/inbound/DebeziumMessageProducerTests.java new file mode 100644 index 0000000000..ded7dc0490 --- /dev/null +++ b/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> debeziumBuilderMock; + + DebeziumEngine> 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(); + } + +} diff --git a/spring-integration-debezium/src/test/java/org/springframework/integration/debezium/it/DebeziumBatchTests.java b/spring-integration-debezium/src/test/java/org/springframework/integration/debezium/it/DebeziumBatchTests.java new file mode 100644 index 0000000000..9601cdeef5 --- /dev/null +++ b/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> payload = (List>) message.getPayload(); + assertThat(payload).hasSize(52); + + for (int i = 0; i < 52; i++) { + ChangeEvent changeEvent = payload.get(i); + + List 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> debeziumEngineBuilder) { + + DebeziumMessageProducer debeziumMessageProducer = new DebeziumMessageProducer(debeziumEngineBuilder); + debeziumMessageProducer.setEnableBatch(true); + debeziumMessageProducer.setOutputChannel(debeziumInputChannel); + return debeziumMessageProducer; + } + + } + +} diff --git a/spring-integration-debezium/src/test/java/org/springframework/integration/debezium/it/DebeziumMySqlTestContainer.java b/spring-integration-debezium/src/test/java/org/springframework/integration/debezium/it/DebeziumMySqlTestContainer.java new file mode 100644 index 0000000000..0b81bd1e1f --- /dev/null +++ b/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); + } + +} diff --git a/spring-integration-debezium/src/test/java/org/springframework/integration/debezium/it/DebeziumStreamTests.java b/spring-integration-debezium/src/test/java/org/springframework/integration/debezium/it/DebeziumStreamTests.java new file mode 100644 index 0000000000..cc3bb3616c --- /dev/null +++ b/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> 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; + } + + } + +} diff --git a/spring-integration-debezium/src/test/java/org/springframework/integration/debezium/it/DebeziumTestConfiguration.java b/spring-integration-debezium/src/test/java/org/springframework/integration/debezium/it/DebeziumTestConfiguration.java new file mode 100644 index 0000000000..bc2c4c5170 --- /dev/null +++ b/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> 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 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(); + } + +} diff --git a/spring-integration-debezium/src/test/java/org/springframework/integration/debezium/support/DefaultDebeziumHeaderMapperTests.java b/spring-integration-debezium/src/test/java/org/springframework/integration/debezium/support/DefaultDebeziumHeaderMapperTests.java new file mode 100644 index 0000000000..64e2dbee5e --- /dev/null +++ b/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> 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(String key, T value) implements Header { + + @Override + public String getKey() { + return key; + } + + @Override + public T getValue() { + return value; + } + + } + +} diff --git a/spring-integration-debezium/src/test/resources/log4j2-test.xml b/spring-integration-debezium/src/test/resources/log4j2-test.xml new file mode 100644 index 0000000000..fefc341a07 --- /dev/null +++ b/spring-integration-debezium/src/test/resources/log4j2-test.xml @@ -0,0 +1,16 @@ + + + + + + + + + + + + + + + + diff --git a/src/reference/asciidoc/debezium.adoc b/src/reference/asciidoc/debezium.adoc new file mode 100644 index 0000000000..a32e9d670a --- /dev/null +++ b/src/reference/asciidoc/debezium.adoc @@ -0,0 +1,141 @@ +[[debezium]] +== Debezium Support + +Spring Integration provides channel adapter for handling Change Events using Debezium. + +https://debezium.io/documentation/reference/development/engine.html[Debezium Engine] based Change Data Capture (CDC) channel adapter. +The Debezium adapter allows capturing database change events, converting them into messages and streaming those to the outbound channels. + +You need to include this dependency into your project: + +==== +[source, xml, subs="normal", role="primary"] +.Maven +---- + + org.springframework.integration + spring-integration-debezium + {project-version} + +---- +[source, groovy, subs="normal", role="secondary"] +.Gradle +---- +compile "org.springframework.integration:spring-integration-debezium:{project-version}" +---- +==== + +[[debezium-inbound]] +=== Inbound Debezium Channel Adapter + +The Debezium adapter expects a pre-configured `DebeziumEngine.Builder>` bean instance. +Additionally, the `DebeziumMessageProducer` can be tuned with the following configuration properties: + +- `contentType` - allows handling for `JSON` (default), `AVRO` and `PROTOBUF` message contents. +The contentType `must` be be aligned with the `SerializationFormat` configured for the provided `DebeziumEngine.Builder`. +- `enableBatch` - when set to `false` (default), the debezium adapter would send new `Message` for every `ChangeEvent` data change event received from the source database. +If set to `true` then the adapter sends downstream a single `Message` for each batch of `ChangeEvent` received from the Debezium engine. +Such a payload is not serializable and would require a custom serialization/deserialization implementation. +- `enableEmptyPayload` - Enables 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 `Optional.empty`. +Defaults to `false`. +- `headerMapper` - custom `HeaderMapper` implementation that allows for selecting and converting the `ChangeEvent` headers into `Message` headers. +The default `DefaultDebeziumHeaderMapper` implementation (no headers are mapped) provides a setter for `setHeaderNamesToMap`. +- `threadFactory` - Set custom `ThreadFactory` for the Debezium executor service. +Debezium Engine is designed to be submitted to an `Executor` or `ExecutorService` for execution by single thread. + +The following code snippets demonstrate various configuration for this channel adapter: + +==== Configuring with Java Configuration + +The following Spring Boot application shows an example of how to configure the inbound adapter with Java configuration: + +==== +[source, java] +---- +@SpringBootApplication +public class DebeziumJavaApplication { + + public static void main(String[] args) { + new SpringApplicationBuilder(DebeziumJavaApplication.class) + .web(false) + .run(args); + } + + @Bean + public MessageChannel debeziumInputChannel() { + return new DirectChannel(); + } + + @Bean + public MessageProducer debeziumMessageProducer( + DebeziumEngine.Builder> debeziumEngineBuilder, + MessageChannel debeziumInputChannel) { + + DebeziumMessageProducer debeziumMessageProducer = + new DebeziumMessageProducer(debeziumEngineBuilder); + debeziumMessageProducer.setOutputChannel(debeziumInputChannel); + return debeziumMessageProducer; + } + + @ServiceActivator(inputChannel = "debeziumInputChannel") + public void handler(String changeEventData) { + System.out.println(changeEventData); + } + +} +---- +==== + +Similarly, we can configure the `DebeziumMessageProducer` to process the incoming change events in batches: + +==== +[source, java] +---- +@Bean +public MessageProducer debeziumMessageProducer( + DebeziumEngine.Builder> debeziumEngineBuilder, + MessageChannel debeziumInputChannel) { + + DebeziumMessageProducer debeziumMessageProducer = new DebeziumMessageProducer(debeziumEngineBuilder); + debeziumMessageProducer.setEnableBatch(true); + debeziumMessageProducer.setOutputChannel(debeziumInputChannel); + return debeziumMessageProducer; +} + +@ServiceActivator(inputChannel = "debeziumInputChannel") +public void handler(List> payload) { + System.out.println(payload); +} +---- +==== + + +==== Configuring with the Java DSL + +The following Spring Boot application provides an example of configuring the inbound adapter with the Java DSL: + +==== +[source, java] +---- +@SpringBootApplication +public class DebeziumJavaApplication { + + public static void main(String[] args) { + new SpringApplicationBuilder(DebeziumJavaApplication.class) + .web(false) + .run(args); + } + + @Bean + public IntegrationFlow debeziumInbound( + DebeziumEngine.Builder> debeziumEngineBuilder) { + + return IntegrationFlow.from(new DebeziumMessageProducer(debeziumEngineBuilder)) + .handle(m -> System.out.println(new String((byte[]) m.getPayload()))) + .get(); + } + +} +---- +==== diff --git a/src/reference/asciidoc/endpoint-summary.adoc b/src/reference/asciidoc/endpoint-summary.adoc index e7eb365ccd..c60ca18b5b 100644 --- a/src/reference/asciidoc/endpoint-summary.adoc +++ b/src/reference/asciidoc/endpoint-summary.adoc @@ -66,6 +66,12 @@ The following table summarizes the various endpoints with quick links to the app | N | <<./cassandra.adoc#cassandra-outbound,Outbound Gateway>> +| *Debezium* +| <<./debezium.adoc#debezium-inbound, Debezium Inbound Channel Adapter>> +| N +| N +| N + | *Events* | <<./event.adoc#appevent-inbound,Receiving Spring Application Events>> | <<./event.adoc#appevent-outbound,Sending Spring Application Events>> diff --git a/src/reference/asciidoc/index-single.adoc b/src/reference/asciidoc/index-single.adoc index 319ea5f233..f2d40b1943 100644 --- a/src/reference/asciidoc/index-single.adoc +++ b/src/reference/asciidoc/index-single.adoc @@ -51,6 +51,8 @@ include::./camel.adoc[] include::./cassandra.adoc[] +include::./debezium.adoc[] + include::./event.adoc[] include::./feed.adoc[] diff --git a/src/reference/asciidoc/index.adoc b/src/reference/asciidoc/index.adoc index 2bb299ea59..14bdf8576c 100644 --- a/src/reference/asciidoc/index.adoc +++ b/src/reference/asciidoc/index.adoc @@ -38,6 +38,7 @@ Welcome to the Spring Integration reference documentation! <<./amqp.adoc#amqp,AMQP Support>> :: AMQP channels, adapters and gateways <<./camel.adoc#camel,Apache Camel Support>> :: Apache Camel channel adapters and gateways <<./cassandra.adoc#cassandra,Apache Cassandra Support>> :: Apache Cassandra channel adapters +<<./debezium.adoc#debezium,Debezium CDC Support>> :: Debezium channel adapters <<./event.adoc#applicationevent,Spring `ApplicationEvent` Support>> :: Handling and consuming Spring application events with channel adapters <<./feed.adoc#feed,Feed Adapter>> :: RSS and Atom channel adapters <<./file.adoc#files,File Support>> :: Channel adapters and gateways for file system support diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 69c9a353e1..155708b503 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -17,6 +17,12 @@ In general the project has been moved to the latest dependency versions. [[x6.2-new-components]] === New Components +[[x6.2-debezium]] +==== Debezium Inbound Channel Adapter + +The Debezium Engine based Change Data Capture (CDC) channel adapter, that allows capturing database change events, converting them into Messages and streaming those to the outbound channels. +See <<./debezium.adoc#debezium-inbound, Debezium Inbound Channel Adapter>> for more information. + [[x6.2-general]] === General Changes