diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-common/pom.xml b/flink-end-to-end-tests/flink-end-to-end-tests-common/pom.xml
index 35f7e14799cc7..0c9e44b3d245b 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-common/pom.xml
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-common/pom.xml
@@ -69,6 +69,13 @@ under the License.
 			<groupId>org.testcontainers</groupId>
 			<artifactId>testcontainers</artifactId>
 		</dependency>
+		<dependency>
+			<!-- Testcontainers 2.x no longer pulls in JUnit 4 transitively,
+				but this module's main sources still use it (TemporaryFolder, Assume). -->
+			<groupId>junit</groupId>
+			<artifactId>junit</artifactId>
+			<scope>compile</scope>
+		</dependency>
 		<dependency>
 			<!-- To ensure that flink-dist is built beforehand -->
 			<groupId>org.apache.flink</groupId>
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-hive/src/test/java/org/apache/flink/tests/hive/HiveITCase.java b/flink-end-to-end-tests/flink-end-to-end-tests-hive/src/test/java/org/apache/flink/tests/hive/HiveITCase.java
index 24a759887c572..4248426c1c787 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-hive/src/test/java/org/apache/flink/tests/hive/HiveITCase.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-hive/src/test/java/org/apache/flink/tests/hive/HiveITCase.java
@@ -33,6 +33,7 @@
 import org.apache.flink.util.UserClassLoaderJarTestUtils;
 
 import org.apache.hadoop.hive.conf.HiveConf;
+import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.Rule;
@@ -68,7 +69,8 @@ public class HiveITCase extends TestLogger {
 
     @ClassRule public static final TemporaryFolder TMP_FOLDER = new TemporaryFolder();
 
-    @ClassRule
+    // Testcontainers 2.x containers are no longer JUnit 4 TestRules, so the lifecycle is driven
+    // manually from @BeforeClass/@AfterClass instead of via @ClassRule.
     public static final HiveContainers.HiveContainer HIVE_CONTAINER =
             HiveContainers.createHiveContainer(
                     Arrays.asList("hive_sink1", "hive_sink2", "h_table_sink1", "h_table_sink2"));
@@ -93,10 +95,16 @@ public class HiveITCase extends TestLogger {
 
     @BeforeClass
     public static void beforeClass() throws Exception {
+        HIVE_CONTAINER.start();
         initUDFJar();
         initHiveConfFile();
     }
 
+    @AfterClass
+    public static void afterClass() {
+        HIVE_CONTAINER.stop();
+    }
+
     private static void initUDFJar() throws Exception {
         Path tmpPath = TMP_FOLDER.getRoot().toPath();
         LOG.info("The current temporary path: {}", tmpPath);
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-hive/src/test/java/org/apache/flink/tests/hive/containers/HiveContainers.java b/flink-end-to-end-tests/flink-end-to-end-tests-hive/src/test/java/org/apache/flink/tests/hive/containers/HiveContainers.java
index 74d3692216c1a..d32238b750234 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-hive/src/test/java/org/apache/flink/tests/hive/containers/HiveContainers.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-hive/src/test/java/org/apache/flink/tests/hive/containers/HiveContainers.java
@@ -27,7 +27,6 @@
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
 import okhttp3.Response;
-import org.junit.runner.Description;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.GenericContainer;
@@ -86,9 +85,9 @@ protected void doStart() {
         }
 
         @Override
-        protected void finished(Description description) {
+        protected void containerIsStopping(InspectContainerResponse containerInfo) {
             backupLogs();
-            super.finished(description);
+            super.containerIsStopping(containerInfo);
         }
 
         @Override
diff --git a/flink-end-to-end-tests/flink-sql-client-test/pom.xml b/flink-end-to-end-tests/flink-sql-client-test/pom.xml
index 3a69500525907..658400eed975c 100644
--- a/flink-end-to-end-tests/flink-sql-client-test/pom.xml
+++ b/flink-end-to-end-tests/flink-sql-client-test/pom.xml
@@ -73,7 +73,7 @@ under the License.
 		</dependency>
 		<dependency>
 			<groupId>org.testcontainers</groupId>
-			<artifactId>kafka</artifactId>
+			<artifactId>testcontainers-kafka</artifactId>
 			<scope>test</scope>
 		</dependency>
 
diff --git a/flink-end-to-end-tests/flink-sql-client-test/src/test/java/SqlClientITCase.java b/flink-end-to-end-tests/flink-sql-client-test/src/test/java/SqlClientITCase.java
index 7f49a6ea96716..3216ef6024a60 100644
--- a/flink-end-to-end-tests/flink-sql-client-test/src/test/java/SqlClientITCase.java
+++ b/flink-end-to-end-tests/flink-sql-client-test/src/test/java/SqlClientITCase.java
@@ -41,11 +41,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.KafkaContainer;
 import org.testcontainers.containers.Network;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
 import org.testcontainers.junit.jupiter.Container;
 import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.kafka.ConfluentKafkaContainer;
 import org.testcontainers.utility.DockerImageName;
 
 import java.io.File;
@@ -78,10 +78,10 @@ public class SqlClientITCase {
     public static final Network NETWORK = Network.newNetwork();
 
     @Container
-    public static final KafkaContainer KAFKA =
-            new KafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA))
+    public static final ConfluentKafkaContainer KAFKA =
+            new ConfluentKafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA))
                     .withNetwork(NETWORK)
-                    .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS)
+                    .withListener(INTER_CONTAINER_KAFKA_ALIAS + ":19092")
                     .withLogConsumer(LOG_CONSUMER);
 
     public final FlinkContainers flink =
@@ -219,7 +219,7 @@ void testMatchRecognize() throws Exception {
                         "    'topic' = 'test-json',",
                         "    'properties.bootstrap.servers' = '"
                                 + INTER_CONTAINER_KAFKA_ALIAS
-                                + ":9092',",
+                                + ":19092',",
                         "    'scan.startup.mode' = 'earliest-offset',",
                         "    'format' = 'json',",
                         "    'json.timestamp-format.standard' = 'ISO-8601'",
diff --git a/flink-end-to-end-tests/flink-sql-gateway-test/src/test/java/org/apache/flink/table/gateway/SqlGatewayE2ECase.java b/flink-end-to-end-tests/flink-sql-gateway-test/src/test/java/org/apache/flink/table/gateway/SqlGatewayE2ECase.java
index 0d7387b5803fa..854bd5b59b8a4 100644
--- a/flink-end-to-end-tests/flink-sql-gateway-test/src/test/java/org/apache/flink/table/gateway/SqlGatewayE2ECase.java
+++ b/flink-end-to-end-tests/flink-sql-gateway-test/src/test/java/org/apache/flink/table/gateway/SqlGatewayE2ECase.java
@@ -93,7 +93,9 @@ public class SqlGatewayE2ECase extends TestLogger {
     private static final String RESULT_KEY = "$RESULT";
 
     @ClassRule public static final TemporaryFolder FOLDER = new TemporaryFolder();
-    @ClassRule public static final HiveContainer HIVE_CONTAINER = new HiveContainer();
+
+    public static final HiveContainer HIVE_CONTAINER = new HiveContainer();
+
     @Rule public final FlinkResource flinkResource = buildFlinkResource();
 
     private static NetUtils.Port hiveserver2Port;
@@ -107,6 +109,7 @@ public class SqlGatewayE2ECase extends TestLogger {
 
     @BeforeClass
     public static void beforeClass() {
+        HIVE_CONTAINER.start();
         ENDPOINT_CONFIG.setString(
                 getPrefixedConfigOptionName(CATALOG_HIVE_CONF_DIR), createHiveConf().getParent());
     }
@@ -115,6 +118,7 @@ public static void beforeClass() {
     public static void afterClass() throws Exception {
         hiveserver2Port.close();
         restPort.close();
+        HIVE_CONTAINER.stop();
     }
 
     @Test
diff --git a/flink-end-to-end-tests/flink-sql-gateway-test/src/test/java/org/apache/flink/table/gateway/containers/HiveContainer.java b/flink-end-to-end-tests/flink-sql-gateway-test/src/test/java/org/apache/flink/table/gateway/containers/HiveContainer.java
index 6fa9064fcfd73..caa384c63fc70 100644
--- a/flink-end-to-end-tests/flink-sql-gateway-test/src/test/java/org/apache/flink/table/gateway/containers/HiveContainer.java
+++ b/flink-end-to-end-tests/flink-sql-gateway-test/src/test/java/org/apache/flink/table/gateway/containers/HiveContainer.java
@@ -25,7 +25,6 @@
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
 import okhttp3.Response;
-import org.junit.runner.Description;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.GenericContainer;
@@ -96,9 +95,9 @@ protected void containerIsStarted(InspectContainerResponse containerInfo) {
     }
 
     @Override
-    protected void finished(Description description) {
+    protected void containerIsStopping(InspectContainerResponse containerInfo) {
         backupLogs();
-        super.finished(description);
+        super.containerIsStopping(containerInfo);
     }
 
     public String getHiveMetastoreURI() {
diff --git a/flink-end-to-end-tests/test-scripts/test_confluent_schema_registry.sh b/flink-end-to-end-tests/test-scripts/test_confluent_schema_registry.sh
index 324e3a34c5526..d653ef46bc578 100755
--- a/flink-end-to-end-tests/test-scripts/test_confluent_schema_registry.sh
+++ b/flink-end-to-end-tests/test-scripts/test_confluent_schema_registry.sh
@@ -20,8 +20,8 @@
 set -Eeuo pipefail
 
 KAFKA_VERSION="3.2.3"
-CONFLUENT_VERSION="7.2.9"
-CONFLUENT_MAJOR_VERSION="7.2"
+CONFLUENT_VERSION="7.5.3"
+CONFLUENT_MAJOR_VERSION="7.5"
 # Check the Confluent Platform <> Apache Kafka compatibility matrix when updating KAFKA_VERSION
 KAFKA_SQL_VERSION="universal"
 
diff --git a/flink-end-to-end-tests/test-scripts/test_pyflink.sh b/flink-end-to-end-tests/test-scripts/test_pyflink.sh
index 2b53e3393bb8d..34654d3d3a286 100755
--- a/flink-end-to-end-tests/test-scripts/test_pyflink.sh
+++ b/flink-end-to-end-tests/test-scripts/test_pyflink.sh
@@ -20,8 +20,8 @@
 set -Eeuo pipefail
 
 KAFKA_VERSION="3.2.3"
-CONFLUENT_VERSION="7.2.9"
-CONFLUENT_MAJOR_VERSION="7.2"
+CONFLUENT_VERSION="7.5.3"
+CONFLUENT_MAJOR_VERSION="7.5"
 # Check the Confluent Platform <> Apache Kafka compatibility matrix when updating KAFKA_VERSION
 KAFKA_SQL_VERSION="universal"
 SQL_JARS_DIR=${END_TO_END_DIR}/flink-sql-client-test/target/sql-jars
diff --git a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/util/DockerImageVersions.java b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/util/DockerImageVersions.java
index c48ad4d8525fa..f781262669c08 100644
--- a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/util/DockerImageVersions.java
+++ b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/util/DockerImageVersions.java
@@ -28,9 +28,9 @@
  */
 public class DockerImageVersions {
 
-    public static final String KAFKA = "confluentinc/cp-kafka:7.2.9";
+    public static final String KAFKA = "confluentinc/cp-kafka:7.5.3";
 
-    public static final String SCHEMA_REGISTRY = "confluentinc/cp-schema-registry:7.2.9";
+    public static final String SCHEMA_REGISTRY = "confluentinc/cp-schema-registry:7.5.3";
 
     public static final String KINESALITE = "instructure/kinesalite:latest";
 
diff --git a/pom.xml b/pom.xml
index 60b5f6ed51caa..b2cfd2fe2fa9a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -162,7 +162,7 @@ under the License.
 		<beam.version>2.43.0</beam.version>
 		<protoc.version>3.21.7</protoc.version>
 		<okhttp.version>3.14.9</okhttp.version>
-		<testcontainers.version>1.21.4</testcontainers.version>
+		<testcontainers.version>2.0.5</testcontainers.version>
 		<lz4.version>1.10.3</lz4.version>
 		<commons.io.version>2.15.1</commons.io.version>
 		<japicmp.skip>false</japicmp.skip>
@@ -286,7 +286,7 @@ under the License.
 
 		<dependency>
 			<groupId>org.testcontainers</groupId>
-			<artifactId>junit-jupiter</artifactId>
+			<artifactId>testcontainers-junit-jupiter</artifactId>
 			<scope>test</scope>
 		</dependency>