diff --git a/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineAutoConfiguration.java b/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineBuilderAutoConfiguration.java similarity index 55% rename from supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineAutoConfiguration.java rename to supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineBuilderAutoConfiguration.java index 3b1dfa09..dc42bba9 100644 --- a/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineAutoConfiguration.java +++ b/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineBuilderAutoConfiguration.java @@ -18,52 +18,62 @@ package org.springframework.cloud.fn.supplier.debezium; import java.time.Clock; import java.time.Duration; -import java.util.Map; import java.util.Objects; -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.CompletionCallback; import io.debezium.engine.DebeziumEngine.ConnectorCallback; +import io.debezium.engine.format.KeyValueHeaderChangeEventFormat; import io.debezium.engine.format.SerializationFormat; import io.debezium.engine.spi.OffsetCommitPolicy; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.boot.autoconfigure.AutoConfiguration; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.autoconfigure.condition.AnyNestedCondition; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.supplier.debezium.DebeziumProperties.DebeziumFormat; import org.springframework.context.annotation.Bean; - -import static org.springframework.cloud.fn.supplier.debezium.DebeziumProperties.DebeziumFormat; +import org.springframework.context.annotation.Conditional; +import org.springframework.core.Ordered; +import org.springframework.core.annotation.Order; /** - * DebeziumEngine auto-configuration. - * - * The engine configuration is entirely standalone and only talks with the source system; Applications using the engine - * auto-configuration simply provides a {@link Consumer consumer function} implementation to which the engine will pass - * all records containing database change events. + * {@link EnableAutoConfiguration Auto-configuration} for {@link DebeziumEngine.Builder}. *

- * With the engine, the application that runs the connector assumes all responsibility for fault tolerance, scalability, - * and durability. Additionally, applications must specify how the engine can store its relational database schema - * history and offsets. By default, this information will be stored in memory and will thus be lost upon application - * restart. + * The builder provides a standalone engine configuration that talks with the source data system. + *

+ * The application that runs the debezium engine assumes all responsibility for fault tolerance, scalability, and + * durability. Additionally, applications must specify how the engine can store its relational database schema history + * and offsets. By default, this information will be stored in memory and will thus be lost upon application restart. + *

+ * The {@link DebeziumEngine.Builder} auto-configuration is activated only if a Debezium Connector is available on the + * classpath and the debezium.properties.connector.class property is set. + *

+ * Properties prefixed with debezium.properties are passed through as native Debezium properties. * * @author Christian Tzolov * @author Corneil du Plessis */ @AutoConfiguration @EnableConfigurationProperties(DebeziumProperties.class) -public class DebeziumEngineAutoConfiguration { +@Conditional(DebeziumEngineBuilderAutoConfiguration.OnDebeziumConnectorCondition.class) +@ConditionalOnProperty(prefix = "debezium", name = "properties.connector.class") +public class DebeziumEngineBuilderAutoConfiguration { - private static final Log logger = LogFactory.getLog(DebeziumEngineAutoConfiguration.class); + private static final Log logger = LogFactory.getLog(DebeziumEngineBuilderAutoConfiguration.class); /** * The fully-qualified class name of the commit policy type. The default is a periodic commit policy based upon time * intervals. * @param properties The 'debezium.properties.offset.flush.interval.ms' configuration is compulsory for the Periodic - * policy type. The ALWAYS and DEFAULT doesn't require properties. + * policy type. The ALWAYS and DEFAULT doesn't require additional configuration. */ @Bean @ConditionalOnMissingBean @@ -111,29 +121,44 @@ public class DebeziumEngineAutoConfiguration { public ConnectorCallback connectorCallback() { return DEFAULT_CONNECTOR_CALLBACK; } - private static final Map>> serialFormats = Map.of( - DebeziumFormat.JSON, io.debezium.engine.format.JsonByteArray.class, - DebeziumFormat.AVRO, io.debezium.engine.format.Avro.class, - DebeziumFormat.PROTOBUF, io.debezium.engine.format.Protobuf.class - ); + @Bean - public DebeziumEngine debeziumEngine(Consumer> changeEventConsumer, + public Builder> debeziumEngineBuilder( OffsetCommitPolicy offsetCommitPolicy, CompletionCallback completionCallback, ConnectorCallback connectorCallback, DebeziumProperties properties, Clock debeziumClock) { - Class> format = Objects.requireNonNull(serialFormats.get(properties.getFormat()), "Cannot find format for " + properties.getProperties()); - DebeziumEngine> debeziumEngine = DebeziumEngine - .create(format) + Class> payloadFormat = Objects.requireNonNull( + serializationFormatClass(properties.getPayloadFormat()), + "Cannot find payload format for " + properties.getProperties()); + + Class> headerFormat = Objects.requireNonNull( + serializationFormatClass(properties.getHeaderFormat()), + "Cannot find header format for " + properties.getProperties()); + + return DebeziumEngine + .create(KeyValueHeaderChangeEventFormat.of(payloadFormat, payloadFormat, headerFormat)) .using(properties.getDebeziumNativeConfiguration()) .using(debeziumClock) .using(completionCallback) .using(connectorCallback) - .using((offsetCommitPolicy != NULL_OFFSET_COMMIT_POLICY) ? offsetCommitPolicy : null) - .notifying(changeEventConsumer) - .build(); + .using((offsetCommitPolicy != NULL_OFFSET_COMMIT_POLICY) ? offsetCommitPolicy : null); + } - - return debeziumEngine; + /** + * Converts the {@link DebeziumFormat} enum into Debezium {@link SerializationFormat} class. + * @param debeziumFormat debezium format property. + */ + private Class> serializationFormatClass(DebeziumFormat debeziumFormat) { + switch (debeziumFormat) { + case JSON: + return io.debezium.engine.format.JsonByteArray.class; + case AVRO: + return io.debezium.engine.format.Avro.class; + case PROTOBUF: + return io.debezium.engine.format.Protobuf.class; + default: + throw new IllegalArgumentException("Unknown debezium format: " + debeziumFormat); + } } /** @@ -191,4 +216,57 @@ public class DebeziumEngineAutoConfiguration { throw new UnsupportedOperationException("Unimplemented method 'performCommit'"); } }; + + /** + * Determine if Debezium connector is available. This either kicks in if any debezium connector is available. + */ + @Order(Ordered.LOWEST_PRECEDENCE) + static class OnDebeziumConnectorCondition extends AnyNestedCondition { + + OnDebeziumConnectorCondition() { + super(ConfigurationPhase.REGISTER_BEAN); + } + + @ConditionalOnClass(name = { "io.debezium.connector.mysql.MySqlConnector" }) + static class HasMySqlConnector { + + } + + @ConditionalOnClass(name = "io.debezium.connector.postgresql.PostgresConnector") + static class HasPostgreSqlConnector { + + } + + @ConditionalOnClass(name = "io.debezium.connector.db2.Db2Connector") + static class HasDb2Connector { + + } + + @ConditionalOnClass(name = "io.debezium.connector.oracle.OracleConnector") + static class HasOracleConnector { + + } + + @ConditionalOnClass(name = "io.debezium.connector.sqlserver.SqlServerConnector") + static class HasSqlServerConnector { + + } + + @ConditionalOnClass(name = "io.debezium.connector.mongodb.MongoDbConnector") + static class HasMongoDbConnector { + + } + + @ConditionalOnClass(name = "io.debezium.connector.vitess.VitessConnector") + static class HasVitessConnector { + + } + + @ConditionalOnClass(name = "io.debezium.connector.spanner.SpannerConnector") + static class HasSpannerConnector { + + } + + } + } diff --git a/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumProperties.java b/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumProperties.java index 9e544b06..de625b88 100644 --- a/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumProperties.java +++ b/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumProperties.java @@ -42,14 +42,12 @@ public class DebeziumProperties { */ PROTOBUF("application/x-protobuf"),; - private final String contentType; DebeziumFormat(String contentType) { this.contentType = contentType; } - public final String contentType() { return contentType; } @@ -62,9 +60,14 @@ public class DebeziumProperties { private Map properties = new HashMap<>(); /** - * Change Event message content format. Defaults to 'JSON'. + * {@link ChangeEvent} Key and Payload formats. Defaults to 'JSON'. */ - private DebeziumFormat format = DebeziumFormat.JSON; + private DebeziumFormat payloadFormat = DebeziumFormat.JSON; + + /** + * {@link ChangeEvent} header format. Defaults to 'JSON'. + */ + private DebeziumFormat headerFormat = DebeziumFormat.JSON; /** * Copy Change Event headers into Message headers. @@ -80,12 +83,20 @@ public class DebeziumProperties { return properties; } - public DebeziumFormat getFormat() { - return format; + public DebeziumFormat getPayloadFormat() { + return payloadFormat; } - public void setFormat(DebeziumFormat format) { - this.format = format; + public void setPayloadFormat(DebeziumFormat format) { + this.payloadFormat = format; + } + + public DebeziumFormat getHeaderFormat() { + return headerFormat; + } + + public void setHeaderFormat(DebeziumFormat headerFormat) { + this.headerFormat = headerFormat; } public boolean isCopyHeaders() { @@ -104,8 +115,8 @@ public class DebeziumProperties { ALWAYS, /** * Commits offsets no more than the specified time period. If the specified time is less than {@code 0} then the - * policy will behave as ALWAYS policy. Requires the 'debezium.properties.offset.flush.interval.ms' native property - * to be set. + * policy will behave as ALWAYS policy. Requires the 'debezium.properties.offset.flush.interval.ms' native + * property to be set. */ PERIODIC, /** diff --git a/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumReactiveConsumerConfiguration.java b/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumReactiveConsumerConfiguration.java index fee3e494..bdecf475 100644 --- a/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumReactiveConsumerConfiguration.java +++ b/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumReactiveConsumerConfiguration.java @@ -26,6 +26,7 @@ import java.util.function.Supplier; import io.debezium.engine.ChangeEvent; import io.debezium.engine.DebeziumEngine; +import io.debezium.engine.DebeziumEngine.Builder; import io.debezium.engine.Header; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -85,7 +86,16 @@ public class DebeziumReactiveConsumerConfiguration implements BeanClassLoaderAwa private final ExecutorService debeziumExecutor = Executors.newSingleThreadExecutor(); @Bean - public Supplier>> debeziumSupplier(DebeziumEngine debeziumEngine) { + public DebeziumEngine> debeziumEngine( + Consumer> changeEventConsumer, + Builder> debeziumEngineBuilder) { + + return debeziumEngineBuilder.notifying(changeEventConsumer).build(); + } + + @Bean + public Supplier>> debeziumSupplier(DebeziumEngine> debeziumEngine) { + return () -> this.eventSink.asFlux() .doOnRequest(r -> debeziumExecutor.execute(debeziumEngine)) .doOnTerminate(debeziumExecutor::shutdownNow); @@ -95,7 +105,7 @@ public class DebeziumReactiveConsumerConfiguration implements BeanClassLoaderAwa @ConditionalOnMissingBean public Consumer> changeEventConsumer(DebeziumProperties properties) { - return new ChangeEventConsumer(properties.getFormat().contentType(), properties.isCopyHeaders(), + return new ChangeEventConsumer(properties.getPayloadFormat().contentType(), properties.isCopyHeaders(), this.eventSink); } diff --git a/supplier/debezium-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/supplier/debezium-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports index a85c0324..e47f5f2a 100644 --- a/supplier/debezium-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ b/supplier/debezium-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -1 +1 @@ -org.springframework.cloud.fn.supplier.debezium.DebeziumEngineAutoConfiguration +org.springframework.cloud.fn.supplier.debezium.DebeziumEngineBuilderAutoConfiguration diff --git a/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineBuilderAutoConfigurationTests.java b/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineBuilderAutoConfigurationTests.java new file mode 100644 index 00000000..39f419a3 --- /dev/null +++ b/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineBuilderAutoConfigurationTests.java @@ -0,0 +1,63 @@ +/* + * 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.cloud.fn.supplier.debezium; + +import io.debezium.engine.DebeziumEngine; +import org.junit.jupiter.api.Test; + +import org.springframework.boot.autoconfigure.AutoConfigurations; +import org.springframework.boot.test.context.FilteredClassLoader; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for {@link DebeziumEngineBuilderAutoConfiguration}. + * + * @author Christian Tzolov + */ +public class DebeziumEngineBuilderAutoConfigurationTests { + + private final ApplicationContextRunner contextRunner = new ApplicationContextRunner() + .withConfiguration(AutoConfigurations.of(DebeziumEngineBuilderAutoConfiguration.class)); + + // We have the debezium connectors on the classpath by default. + + @Test + void noConnectorNoProperty() { + this.contextRunner.run((context) -> { + assertThat(context).doesNotHaveBean(DebeziumEngine.Builder.class); + }); + } + + @Test + void noConnectorWithProperty() { + this.contextRunner.withPropertyValues("debezium.properties.connector.class=Dummy") + .withClassLoader(new FilteredClassLoader("io.debezium.connector")) + .run((context) -> { + assertThat(context).doesNotHaveBean(DebeziumEngine.Builder.class); + }); + } + + @Test + void withConnectorWithProperty() { + this.contextRunner.withPropertyValues("debezium.properties.connector.class=Dummy").run((context) -> { + assertThat(context).hasSingleBean(DebeziumEngine.Builder.class); + }); + } + +} diff --git a/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/DebeziumPropertiesTests.java b/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/DebeziumPropertiesTests.java new file mode 100644 index 00000000..b2314c24 --- /dev/null +++ b/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/DebeziumPropertiesTests.java @@ -0,0 +1,60 @@ +/* + * 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.cloud.fn.supplier.debezium; + +import org.junit.jupiter.api.Test; + +import org.springframework.cloud.fn.supplier.debezium.DebeziumProperties.DebeziumFormat; +import org.springframework.cloud.fn.supplier.debezium.DebeziumProperties.DebeziumOffsetCommitPolicy; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for {@link DebeziumProperties}. + * + * @author Christian Tzolov + */ +public class DebeziumPropertiesTests { + + DebeziumProperties properties = new DebeziumProperties(); + + @Test + public void defaultPropertiesTest() { + assertThat(this.properties.getPayloadFormat()).isEqualTo(DebeziumFormat.JSON); + assertThat(this.properties.getHeaderFormat()).isEqualTo(DebeziumFormat.JSON); + assertThat(this.properties.isCopyHeaders()).isEqualTo(true); + assertThat(this.properties.getOffsetCommitPolicy()).isEqualTo(DebeziumOffsetCommitPolicy.DEFAULT); + assertThat(this.properties.getProperties()).isNotNull(); + assertThat(this.properties.getProperties()).isEmpty(); + } + + @Test + public void debeziumFormatTest() { + this.properties.setPayloadFormat(DebeziumFormat.AVRO); + assertThat(this.properties.getPayloadFormat()).isEqualTo(DebeziumFormat.AVRO); + assertThat(this.properties.getPayloadFormat().contentType()).isEqualTo("application/avro"); + + this.properties.setPayloadFormat(DebeziumFormat.JSON); + assertThat(this.properties.getPayloadFormat()).isEqualTo(DebeziumFormat.JSON); + assertThat(this.properties.getPayloadFormat().contentType()).isEqualTo("application/json"); + + this.properties.setPayloadFormat(DebeziumFormat.PROTOBUF); + assertThat(this.properties.getPayloadFormat()).isEqualTo(DebeziumFormat.PROTOBUF); + assertThat(this.properties.getPayloadFormat().contentType()).isEqualTo("application/x-protobuf"); + } + +} diff --git a/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/TestJdbcTemplateConfiguration.java b/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/TestJdbcTemplateConfiguration.java similarity index 96% rename from supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/TestJdbcTemplateConfiguration.java rename to supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/TestJdbcTemplateConfiguration.java index 9ff97e61..74a6841e 100644 --- a/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/TestJdbcTemplateConfiguration.java +++ b/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/TestJdbcTemplateConfiguration.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.fn.supplier.debezium; +package org.springframework.cloud.fn.supplier.debezium.it; import javax.sql.DataSource; diff --git a/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/custom/DebeziumEngineAutoConfigurationTest.java b/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/custom/DebeziumEngineBuilderAutoConfigurationIntegrationTest.java similarity index 91% rename from supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/custom/DebeziumEngineAutoConfigurationTest.java rename to supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/custom/DebeziumEngineBuilderAutoConfigurationIntegrationTest.java index 242c33ae..2397eadc 100644 --- a/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/custom/DebeziumEngineAutoConfigurationTest.java +++ b/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/custom/DebeziumEngineBuilderAutoConfigurationIntegrationTest.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.fn.supplier.debezium.custom; +package org.springframework.cloud.fn.supplier.debezium.it.custom; import java.io.File; import java.time.Duration; @@ -25,7 +25,7 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.function.Consumer; import io.debezium.engine.ChangeEvent; -import io.debezium.engine.DebeziumEngine; +import io.debezium.engine.DebeziumEngine.Builder; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.junit.jupiter.api.Tag; @@ -40,7 +40,7 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration; import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration; import org.springframework.boot.test.context.runner.ApplicationContextRunner; -import org.springframework.cloud.fn.supplier.debezium.TestJdbcTemplateConfiguration; +import org.springframework.cloud.fn.supplier.debezium.it.TestJdbcTemplateConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Import; import org.springframework.jdbc.core.JdbcTemplate; @@ -57,8 +57,8 @@ import static org.awaitility.Awaitility.await; */ @Tag("integration") @Testcontainers -public class DebeziumEngineAutoConfigurationTest { - private static final Log logger = LogFactory.getLog(DebeziumEngineAutoConfigurationTest.class); +public class DebeziumEngineBuilderAutoConfigurationIntegrationTest { + private static final Log logger = LogFactory.getLog(DebeziumEngineBuilderAutoConfigurationIntegrationTest.class); private static final String DATABASE_NAME = "inventory"; public static final String IMAGE_TAG = "2.2.0.Final"; @@ -144,8 +144,9 @@ public class DebeziumEngineAutoConfigurationTest { public static class DebeziumCustomConsumerApplication { @Bean - public EmbeddedEngineExecutorService embeddedEngine(DebeziumEngine debeziumEngine) { - return new EmbeddedEngineExecutorService(debeziumEngine); + public EmbeddedEngineExecutorService embeddedEngine(Consumer> changeEventConsumer, + Builder> debeziumEngineBuilder) { + return new EmbeddedEngineExecutorService(debeziumEngineBuilder.notifying(changeEventConsumer).build()); } @Bean diff --git a/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/custom/EmbeddedEngineExecutorService.java b/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/custom/EmbeddedEngineExecutorService.java similarity index 97% rename from supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/custom/EmbeddedEngineExecutorService.java rename to supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/custom/EmbeddedEngineExecutorService.java index ff599384..c370943d 100644 --- a/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/custom/EmbeddedEngineExecutorService.java +++ b/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/custom/EmbeddedEngineExecutorService.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.fn.supplier.debezium.custom; +package org.springframework.cloud.fn.supplier.debezium.it.custom; import java.io.IOException; import java.util.concurrent.ExecutorService; diff --git a/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/supplier/DebeziumSupplierTest.java b/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/supplier/DebeziumSupplierIntegrationTest.java similarity index 97% rename from supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/supplier/DebeziumSupplierTest.java rename to supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/supplier/DebeziumSupplierIntegrationTest.java index 1851875c..4bccf8fa 100644 --- a/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/supplier/DebeziumSupplierTest.java +++ b/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/supplier/DebeziumSupplierIntegrationTest.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.fn.supplier.debezium.supplier; +package org.springframework.cloud.fn.supplier.debezium.it.supplier; import java.util.function.Supplier; @@ -32,7 +32,7 @@ import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.cloud.fn.supplier.debezium.DebeziumReactiveConsumerConfiguration; -import org.springframework.cloud.fn.supplier.debezium.TestJdbcTemplateConfiguration; +import org.springframework.cloud.fn.supplier.debezium.it.TestJdbcTemplateConfiguration; import org.springframework.context.annotation.Import; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.messaging.Message; @@ -76,7 +76,7 @@ import static org.assertj.core.api.Assertions.assertThat; "app.datasource.type=com.zaxxer.hikari.HikariDataSource" }) @Testcontainers -public class DebeziumSupplierTest { +public class DebeziumSupplierIntegrationTest { public static final String IMAGE_TAG = "2.2.0.Final"; public static final String DEBEZIUM_EXAMPLE_MYSQL_IMAGE = "debezium/example-mysql:" + IMAGE_TAG;