[FLINK-39527][test] Migrate flink-end-to-end-tests to JUnit 5

flink-end-to-end-tests/flink-batch-sql-test/src/test/java/org/apache/flink/sql/tests/BatchSQLTest.java

@@ -51,7 +51,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
 
 @ExtendWith(TestLoggerExtension.class)
 class BatchSQLTest {
@@ -79,7 +79,7 @@ class BatchSQLTest {
                 // Only above shuffle modes are supported by the adaptive batch scheduler
                 // , "ALL_EXCHANGES_PIPELINE"
             })
-    public void testBatchSQL(BatchShuffleMode shuffleMode, @TempDir Path tmpDir) throws Exception {
+    void testBatchSQL(BatchShuffleMode shuffleMode, @TempDir Path tmpDir) throws Exception {
         LOG.info("Results for this test will be stored at: {}", tmpDir);
 
         String sqlStatement = new String(Files.readAllBytes(sqlPath));
@@ -143,13 +143,13 @@ public void testBatchSQL(BatchShuffleMode shuffleMode, @TempDir Path tmpDir) thr
                         .filter(file -> !file.isDirectory())
                         .map(File::getPath)
                         .collect(Collectors.toList());
-        assertEquals(1, files.size());
+        assertThat(files).hasSize(1);
         Path resultFile = Paths.get(files.get(0));
 
         LOG.info("Result found at {}", resultFile);
         String actual = new String(Files.readAllBytes(resultFile));
         LOG.info("Actual result is: '{}'", actual);
 
-        assertEquals(expected, actual);
+        assertThat(actual).isEqualTo(expected);
     }
 }

flink-end-to-end-tests/flink-datastream-allround-test/src/test/java/org/apache/flink/streaming/tests/AllroundMiniClusterTest.java

@@ -21,39 +21,41 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.StateRecoveryOptions;
 import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.util.TestLoggerExtension;
 
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
-import org.junit.Ignore;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
 
-import java.io.File;
+import java.nio.file.Path;
 
 import static org.apache.flink.configuration.JobManagerOptions.EXECUTION_FAILOVER_STRATEGY;
 
 /** DataStreamAllroundTestProgram on MiniCluster for manual debugging purposes. */
-@Ignore("Test is already part of end-to-end tests. This is for manual debugging.")
-public class AllroundMiniClusterTest extends TestLogger {
+@Disabled("Test is already part of end-to-end tests. This is for manual debugging.")
+@ExtendWith(TestLoggerExtension.class)
+class AllroundMiniClusterTest {
 
-    @BeforeClass
-    public static void beforeClass() {
+    @BeforeAll
+    static void beforeClass() {
         org.apache.log4j.PropertyConfigurator.configure(
                 AllroundMiniClusterTest.class.getClassLoader().getResource("log4j.properties"));
     }
 
-    @ClassRule
-    public static MiniClusterWithClientResource miniClusterResource =
-            new MiniClusterWithClientResource(
+    @RegisterExtension
+    private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+            new MiniClusterExtension(
                     new MiniClusterResourceConfiguration.Builder()
                             .setNumberTaskManagers(4)
                             .setNumberSlotsPerTaskManager(2)
                             .setConfiguration(createConfiguration())
                             .build());
 
-    @ClassRule public static TemporaryFolder temporaryFolder = new TemporaryFolder();
+    @TempDir private static Path temporaryFolder;
 
     private static Configuration createConfiguration() {
         Configuration configuration = new Configuration();
@@ -63,12 +65,13 @@ private static Configuration createConfiguration() {
     }
 
     @Test
-    public void runTest() throws Exception {
-        File checkpointDir = temporaryFolder.newFolder();
+    void runTest() throws Exception {
+        Path checkpointDir = temporaryFolder.resolve("checkpoints");
+        java.nio.file.Files.createDirectories(checkpointDir);
         DataStreamAllroundTestProgram.main(
                 new String[] {
                     "--environment.parallelism", "8",
-                    "--state_backend.checkpoint_directory", checkpointDir.toURI().toString(),
+                    "--state_backend.checkpoint_directory", checkpointDir.toUri().toString(),
                     "--state_backend", "rocks",
                     "--state_backend.rocks.incremental", "true",
                     "--test.simulate_failure", "true",

flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/cache/DownloadCache.java

@@ -19,7 +19,6 @@
 package org.apache.flink.tests.util.cache;
 
 import org.apache.flink.tests.util.util.FactoryUtils;
-import org.apache.flink.util.ExternalResource;
 
 import java.io.IOException;
 import java.nio.file.Path;
@@ -30,7 +29,15 @@
  *
  * <p>Whether, how, and for how long files are cached is implementation-dependent.
  */
-public interface DownloadCache extends ExternalResource {
+public interface DownloadCache {
+
+    void before() throws Exception;
+
+    void afterTestSuccess();
+
+    default void afterTestFailure() {
+        afterTestSuccess();
+    }
 
     /**
      * Returns either a cached or newly downloaded version of the given file. The returned file path

flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/cache/DownloadCacheExtension.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *     http://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.apache.flink.tests.util.cache;
+
+import org.junit.jupiter.api.extension.AfterEachCallback;
+import org.junit.jupiter.api.extension.BeforeEachCallback;
+import org.junit.jupiter.api.extension.ExtensionContext;
+
+import java.io.IOException;
+import java.nio.file.Path;
+
+/** JUnit 5 extension that wraps a {@link DownloadCache} and manages its lifecycle. */
+public class DownloadCacheExtension implements BeforeEachCallback, AfterEachCallback {
+
+    private final DownloadCache delegate;
+
+    public DownloadCacheExtension(DownloadCache delegate) {
+        this.delegate = delegate;
+    }
+
+    public Path getOrDownload(String url, Path targetDir) throws IOException {
+        return delegate.getOrDownload(url, targetDir);
+    }
+
+    @Override
+    public void beforeEach(ExtensionContext context) throws Exception {
+        delegate.before();
+    }
+
+    @Override
+    public void afterEach(ExtensionContext context) {
+        if (context.getExecutionException().isPresent()) {
+            delegate.afterTestFailure();
+        } else {
+            delegate.afterTestSuccess();
+        }
+    }
+}

flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkResource.java

@@ -20,7 +20,6 @@
 
 import org.apache.flink.test.util.JobSubmission;
 import org.apache.flink.tests.util.util.FactoryUtils;
-import org.apache.flink.util.ExternalResource;
 
 import java.io.IOException;
 import java.time.Duration;
@@ -30,7 +29,15 @@
 import java.util.stream.Stream;
 
 /** Generic interface for interacting with Flink. */
-public interface FlinkResource extends ExternalResource {
+public interface FlinkResource {
+
+    void before() throws Exception;
+
+    void afterTestSuccess();
+
+    default void afterTestFailure() {
+        afterTestSuccess();
+    }
 
     /**
      * Starts a cluster.

flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkResourceExtension.java

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *     http://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.apache.flink.tests.util.flink;
+
+import org.junit.jupiter.api.extension.AfterEachCallback;
+import org.junit.jupiter.api.extension.BeforeEachCallback;
+import org.junit.jupiter.api.extension.ExtensionContext;
+
+/** JUnit 5 extension that wraps a {@link FlinkResource} and manages its lifecycle. */
+public class FlinkResourceExtension implements BeforeEachCallback, AfterEachCallback {
+
+    private final FlinkResource delegate;
+
+    public FlinkResourceExtension(FlinkResource delegate) {
+        this.delegate = delegate;
+    }
+
+    public FlinkResource getFlinkResource() {
+        return delegate;
+    }
+
+    @Override
+    public void beforeEach(ExtensionContext context) throws Exception {
+        delegate.before();
+    }
+
+    @Override
+    public void afterEach(ExtensionContext context) {
+        if (context.getExecutionException().isPresent()) {
+            delegate.afterTestFailure();
+        } else {
+            delegate.afterTestSuccess();
+        }
+    }
+}

flink-end-to-end-tests/flink-end-to-end-tests-common/src/test/java/org/apache/flink/tests/util/TestUtilsTest.java

@@ -19,38 +19,40 @@
 
 import org.apache.flink.tests.util.activation.OperatingSystemRestriction;
 import org.apache.flink.util.OperatingSystem;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
 
-import org.junit.Assert;
-import org.junit.BeforeClass;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
 
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 /** Tests for {@link TestUtils}. */
-public class TestUtilsTest extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class TestUtilsTest {
 
-    @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+    @TempDir private Path temporaryFolder;
 
-    @BeforeClass
-    public static void setupClass() {
+    @BeforeAll
+    static void setupClass() {
         OperatingSystemRestriction.forbid(
                 "Symbolic links usually require special permissions on Windows.",
                 OperatingSystem.WINDOWS);
     }
 
     @Test
-    public void copyDirectory() throws IOException {
+    void copyDirectory() throws IOException {
         Path[] files = {
             Paths.get("file1"), Paths.get("dir1", "file2"),
         };
 
-        Path source = temporaryFolder.newFolder("source").toPath();
+        Path source = Files.createDirectory(temporaryFolder.resolve("source"));
         for (Path file : files) {
             Files.createDirectories(source.resolve(file).getParent());
             Files.createFile(source.resolve(file));
@@ -63,7 +65,7 @@ public void copyDirectory() throws IOException {
         TestUtils.copyDirectory(symbolicLink, target);
 
         for (Path file : files) {
-            Assert.assertTrue(Files.exists(target.resolve(file)));
+            assertThat(target.resolve(file)).exists();
         }
     }
 }