Factor out the Debezium Engine Builder auto-configuration (#459)

* Factor out the Debezium Engine Builder auto-configuration

 - Rename DebeziumEngineAutoConfiguration into DebeziumEngineBuilderAutoConfiguration.
 - Make the DebeziumEngineBuilderAutoConfiguration auto-create a DebeizumEngine.Builder instance.
 - Move the DebeizumEngine bean creation to the DebeziumReactiveConsumerConfiguration.
 - Make DebeziumEngineBuilderAutoConfiguration conditional on Debeizum connecgtor on the class path and debezium.properties.connector.class property set.
 - Minor improvment on the serialization format class resolution.
 - Add support for debezium header serialization format, required for the change event headers to work.

 Resolves #458

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}.
  * <p>
- * 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.
+ * <p>
+ * 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.
+ * <p>
+ * The {@link DebeziumEngine.Builder} auto-configuration is activated only if a Debezium Connector is available on the
+ * classpath and the <code>debezium.properties.connector.class</code> property is set.
+ * <p>
+ * Properties prefixed with <code>debezium.properties</code> 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<DebeziumFormat, Class<? extends SerializationFormat<byte[]>>> 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<ChangeEvent<byte[], byte[]>> changeEventConsumer,
+	public Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder(
 			OffsetCommitPolicy offsetCommitPolicy, CompletionCallback completionCallback,
 			ConnectorCallback connectorCallback, DebeziumProperties properties, Clock debeziumClock) {
 
-		Class<? extends SerializationFormat<byte[]>> format = Objects.requireNonNull(serialFormats.get(properties.getFormat()), "Cannot find format for " + properties.getProperties());
-		DebeziumEngine<ChangeEvent<byte[], byte[]>> debeziumEngine = DebeziumEngine
-				.create(format)
+		Class<? extends SerializationFormat<byte[]>> payloadFormat = Objects.requireNonNull(
+				serializationFormatClass(properties.getPayloadFormat()),
+				"Cannot find payload format for " + properties.getProperties());
+
+		Class<? extends SerializationFormat<byte[]>> 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<? extends SerializationFormat<byte[]>> 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 {
+
+		}
+
+	}
+
 }

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<String, String> 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,
 		/**

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<Flux<Message<?>>> debeziumSupplier(DebeziumEngine<?> debeziumEngine) {
+	public DebeziumEngine<ChangeEvent<byte[], byte[]>> debeziumEngine(
+			Consumer<ChangeEvent<byte[], byte[]>> changeEventConsumer,
+			Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder) {
+
+		return debeziumEngineBuilder.notifying(changeEventConsumer).build();
+	}
+
+	@Bean
+	public Supplier<Flux<Message<?>>> debeziumSupplier(DebeziumEngine<ChangeEvent<byte[], byte[]>> 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<ChangeEvent<byte[], byte[]>> changeEventConsumer(DebeziumProperties properties) {
 
-		return new ChangeEventConsumer<byte[]>(properties.getFormat().contentType(), properties.isCopyHeaders(),
+		return new ChangeEventConsumer<byte[]>(properties.getPayloadFormat().contentType(), properties.isCopyHeaders(),
 				this.eventSink);
 	}
 

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

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);
+		});
+	}
+
+}

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");
+	}
+
+}

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;
 

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<ChangeEvent<byte[], byte[]>> changeEventConsumer,
+				Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder) {
+			return new EmbeddedEngineExecutorService(debeziumEngineBuilder.notifying(changeEventConsumer).build());
 		}
 
 		@Bean

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;

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;