DBZ-7185 Refactor debezium server redis tests

debezium-server-redis/pom.xml

@@ -11,6 +11,13 @@
     <name>Debezium Server Redis Sink Adapter</name>
     <packaging>jar</packaging>
 
+    <properties>
+        <quarkus.container-image.build>true</quarkus.container-image.build>
+        <quarkus.jib.jvm-entrypoint>java</quarkus.jib.jvm-entrypoint>
+        <quarkus.container-image.group>debezium-server-test</quarkus.container-image.group>
+        <quarkus.container-image.name>${artifactId}</quarkus.container-image.name>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>io.debezium</groupId>
@@ -28,6 +35,10 @@
         </dependency>
 
         <!-- Testing -->
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-container-image-jib</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-junit5</artifactId>
@@ -143,6 +154,19 @@
                         <test.type>IT</test.type>
                     </systemProperties>
                     <runOrder>${runOrder}</runOrder>
+                    <systemPropertyVariables>
+                        <test.version.debezium>${version.debezium}</test.version.debezium>
+                        <test.server.image.group>${quarkus.container-image.group}</test.server.image.group>
+                        <test.server.image.name>${quarkus.container-image.name}</test.server.image.name>
+                    </systemPropertyVariables>
+                </configuration>
+            </plugin>
+            <!-- Make sure surefire doesn't run integration tests that are ran by failsafe -->
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <configuration>
+                    <excludes>**/*IT.java</excludes>
                 </configuration>
             </plugin>
         </plugins>

debezium-server-redis/src/test/java/io/debezium/server/redis/DebeziumServerConfigBuilder.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.server.redis;
+
+import static io.debezium.server.redis.TestConstants.MYSQL_DATABASE;
+import static io.debezium.server.redis.TestConstants.MYSQL_PASSWORD;
+import static io.debezium.server.redis.TestConstants.MYSQL_PORT;
+import static io.debezium.server.redis.TestConstants.MYSQL_USER;
+import static io.debezium.server.redis.TestConstants.REDIS_PORT;
+import static io.debezium.server.redis.TestConstants.REDIS_SSL_PORT;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DebeziumServerConfigBuilder {
+    private final Map<String, String> config = new HashMap<>();
+
+    public DebeziumServerConfigBuilder withValue(String key, String value) {
+        config.put(key, value);
+        return this;
+    }
+
+    public List<String> build() {
+        return config
+                .entrySet()
+                .stream()
+                .map(e -> e.getKey() + "=" + e.getValue())
+                .collect(Collectors.toList());
+    }
+
+    public static Map<String, String> baseRedisConfig(DebeziumTestContainerWrapper redis) {
+        return Map.of(
+                "debezium.sink.type", "redis",
+                "debezium.sink.redis.address", redis.getContainerAddress() + ":" + REDIS_PORT);
+    }
+
+    public static Map<String, String> baseSslRedisConfig(DebeziumTestContainerWrapper redis) {
+        return Map.of(
+                "debezium.sink.type", "redis",
+                "debezium.sink.redis.address", redis.getContainerAddress() + ":" + REDIS_SSL_PORT,
+                "debezium.sink.redis.ssl.enabled", "true",
+                "debezium.source.database.ssl.mode", "disabled",
+                "debezium.source.offset.storage", "io.debezium.server.redis.RedisOffsetBackingStore");
+    }
+
+    public Map<String, String> baseMySqlConfig(DebeziumTestContainerWrapper mysql) {
+        Map<String, String> result = new HashMap<>(Map.of("debezium.source.connector.class", "io.debezium.connector.mysql.MySqlConnector",
+                "debezium.source.offset.flush.interval.ms", "0",
+                "debezium.source.topic.prefix", "testc",
+                "debezium.source.database.dbname", MYSQL_DATABASE,
+                "debezium.source.database.hostname", String.valueOf(mysql.getContainerAddress()),
+                "debezium.source.database.port", String.valueOf(MYSQL_PORT),
+                "debezium.source.database.user", MYSQL_USER,
+                "debezium.source.database.password", MYSQL_PASSWORD,
+                "debezium.source.database.server.id", "1",
+                "debezium.source.schema.history.internal", "io.debezium.server.redis.RedisSchemaHistory"));
+        result.put("debezium.source.offset.storage.file.filename", "offset.dat");
+        return result;
+    }
+
+    public DebeziumServerConfigBuilder withBaseMySqlRedisConfig(DebeziumTestContainerWrapper redis, DebeziumTestContainerWrapper mysql) {
+        config.putAll(baseRedisConfig(redis));
+        config.putAll(baseMySqlConfig(mysql));
+        return this;
+    }
+
+    public DebeziumServerConfigBuilder withBaseMysqlSslRedisConfig(DebeziumTestContainerWrapper redis, DebeziumTestContainerWrapper mysql) {
+        config.putAll(baseSslRedisConfig(redis));
+        config.putAll(baseMySqlConfig(mysql));
+        return this;
+    }
+}

debezium-server-redis/src/test/java/io/debezium/server/redis/DebeziumTestContainerWrapper.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.server.redis;
+
+import static org.awaitility.Awaitility.await;
+import static org.testcontainers.containers.output.OutputFrame.OutputType.STDOUT;
+
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.OutputFrame;
+
+import lombok.NonNull;
+
+public class DebeziumTestContainerWrapper extends GenericContainer<DebeziumTestContainerWrapper> {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(DebeziumTestContainerWrapper.class);
+
+    private String networkAlias;
+
+    public DebeziumTestContainerWrapper(@NonNull String dockerImageName) {
+        super(dockerImageName);
+    }
+
+    public void pause() {
+        LOGGER.info("Pausing container " + getContainerName());
+        getDockerClient().pauseContainerCmd(getContainerId()).exec();
+    }
+
+    public void unpause() {
+        LOGGER.info("Unpausing container " + getContainerName());
+        getDockerClient().unpauseContainerCmd(getContainerId()).exec();
+    }
+
+    public DebeziumTestContainerWrapper withNetworkAlias(String alias) {
+        this.networkAlias = alias;
+        return super.withNetworkAliases(alias);
+    }
+
+    public String getContainerAddress() {
+        return networkAlias;
+    }
+
+    public String getStandardOutput() {
+        return getLogs(STDOUT);
+    }
+
+    public void waitForContainerLog(String log) {
+        await()
+                .atMost(60, TimeUnit.SECONDS)
+                .until(() -> getLogs(OutputFrame.OutputType.STDOUT).contains(log));
+    }
+
+    public void waitForStop() {
+        await()
+                .atMost(60, TimeUnit.SECONDS)
+                .until(() -> !isRunning());
+    }
+}