[hotfix] Fix deprecated methods for containers (#3706)

apache · Jul 9, 2024 · be474d6 · be474d6
1 parent 7c423f0
commit be474d6
Show file tree

Hide file tree

Showing 6 changed files with 14 additions and 12 deletions.
diff --git a/...ink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java b/...ink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java
@@ -202,7 +202,7 @@ private Map<String, TopicDescription> describeExternalTopics() {
                             .filter(listing -> !listing.isInternal())
                             .map(TopicListing::name)
                             .collect(Collectors.toList());
-            return adminClient.describeTopics(topics).all().get();
+            return adminClient.describeTopics(topics).allTopicNames().get();
         } catch (Exception e) {
             throw new RuntimeException("Failed to list Kafka topics", e);
         }

diff --git a/...-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBContainer.java b/...-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBContainer.java
@@ -185,11 +185,10 @@ public String executeCommandFileInDatabase(
     }
 
     public String getConnectionString() {
-        return String.format(
-                "mongodb://%s:%d", getContainerIpAddress(), getMappedPort(MONGODB_PORT));
+        return String.format("mongodb://%s:%d", getHost(), getMappedPort(MONGODB_PORT));
     }
 
     public String getHostAndPort() {
-        return String.format("%s:%s", getContainerIpAddress(), getMappedPort(MONGODB_PORT));
+        return String.format("%s:%s", getHost(), getMappedPort(MONGODB_PORT));
     }
 }
diff --git a/...imon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlContainer.java b/...imon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlContainer.java
@@ -54,18 +54,18 @@ public MySqlContainer(MySqlVersion version) {
     }
 
     @Override
-    protected Set<Integer> getLivenessCheckPorts() {
+    public Set<Integer> getLivenessCheckPortNumbers() {
         return new HashSet<>(getMappedPort(MYSQL_PORT));
     }
 
     @Override
     protected void configure() {
         optionallyMapResourceParameterAsVolume(
-                MY_CNF_CONFIG_OVERRIDE_PARAM_NAME, "/etc/mysql/", "mysql-default-conf");
+                MY_CNF_CONFIG_OVERRIDE_PARAM_NAME, "/etc/mysql/", "mysql-default-conf", null);
 
         if (parameters.containsKey(SETUP_SQL_PARAM_NAME)) {
             optionallyMapResourceParameterAsVolume(
-                    SETUP_SQL_PARAM_NAME, "/docker-entrypoint-initdb.d/", "N/A");
+                    SETUP_SQL_PARAM_NAME, "/docker-entrypoint-initdb.d/", "N/A", null);
         }
 
         addEnv("MYSQL_DATABASE", databaseName);

diff --git a/...link-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresContainer.java b/...link-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresContainer.java
@@ -34,7 +34,7 @@ public PostgresContainer(DockerImageName dockerImageName) {
     protected void configure() {
         if (parameters.containsKey(SETUP_SQL_PARAM_NAME)) {
             optionallyMapResourceParameterAsVolume(
-                    SETUP_SQL_PARAM_NAME, "/docker-entrypoint-initdb.d/", "N/A");
+                    SETUP_SQL_PARAM_NAME, "/docker-entrypoint-initdb.d/", "N/A", null);
         }
         super.configure();
     }

diff --git a/...k-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionITCaseBase.java b/...k-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionITCaseBase.java
@@ -41,6 +41,7 @@
 import org.testcontainers.containers.Network;
 import org.testcontainers.containers.PulsarContainer;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.utility.DockerImageName;
 
 import java.io.File;
 import java.io.IOException;
@@ -75,6 +76,8 @@ public class PulsarActionITCaseBase extends CdcActionITCaseBase {
     private static final Network NETWORK = Network.newNetwork();
     private static final String INTER_CONTAINER_PULSAR_ALIAS = "pulsar";
 
+    public static final String IMAGE = "apachepulsar/pulsar";
+
     private final ObjectMapper objectMapper = new ObjectMapper();
 
     private PulsarAdmin admin;
@@ -87,7 +90,7 @@ public class PulsarActionITCaseBase extends CdcActionITCaseBase {
 
     private static PulsarContainerExtension createPulsarContainerExtension() {
         PulsarContainerExtension container =
-                new PulsarContainerExtension("3.0.0") {
+                new PulsarContainerExtension(DockerImageName.parse(IMAGE + ":" + "3.0.0")) {
                     @Override
                     protected void doStart() {
                         super.doStart();
@@ -284,8 +287,8 @@ public PulsarSyncDatabaseActionBuilder(Map<String, String> pulsarConfig) {
     /** Pulsar container extension for junit5. */
     private static class PulsarContainerExtension extends PulsarContainer
             implements BeforeAllCallback, AfterAllCallback {
-        private PulsarContainerExtension(String pulsarVersion) {
-            super(pulsarVersion);
+        private PulsarContainerExtension(DockerImageName dockerImageName) {
+            super(dockerImageName);
         }
 
         @Override

diff --git a/...link/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/kafka/KafkaTableTestBase.java b/...link/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/kafka/KafkaTableTestBase.java
@@ -218,7 +218,7 @@ private Map<String, TopicDescription> describeExternalTopics() {
                             .map(TopicListing::name)
                             .collect(Collectors.toList());
 
-            return adminClient.describeTopics(topics).all().get();
+            return adminClient.describeTopics(topics).allTopicNames().get();
         } catch (Exception e) {
             throw new RuntimeException("Failed to list Kafka topics", e);
         }