This is an automated email from the ASF dual-hosted git repository. snuyanzin pushed a commit to branch release-2.4 in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5cd40d54c1640cf978e691658bc9a9361437586e Author: Sergey Nuyanzin <[email protected]> AuthorDate: Thu Oct 29 09:10:03 2026 +0100 [FLINK-41275][e2e] Start SQL e2e test clusters once per class instead of per test --- .../tests/util/flink/FlinkResourceExtension.java | 36 ++++++++++++++-- .../sql/CompileAndExecuteRemotePlanITCase.java | 6 +-- .../org/apache/flink/table/sql/HdfsITCaseBase.java | 48 ++++++++++++---------- .../org/apache/flink/table/sql/SqlITCaseBase.java | 46 +++++++++++++-------- .../flink/table/sql/UsingRemoteJarITCase.java | 7 ++-- 5 files changed, 95 insertions(+), 48 deletions(-) diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkResourceExtension.java b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkResourceExtension.java index 349a67f783f..174bb648910 100644 --- a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkResourceExtension.java +++ b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkResourceExtension.java @@ -18,15 +18,24 @@ package org.apache.flink.tests.util.flink; +import org.junit.jupiter.api.extension.AfterAllCallback; import org.junit.jupiter.api.extension.AfterEachCallback; +import org.junit.jupiter.api.extension.BeforeAllCallback; 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 { +/** + * JUnit 5 extension that wraps a {@link FlinkResource} and manages its lifecycle. Registered on a + * static field, the resource is shared by all tests of the class; otherwise it is per test. + */ +public class FlinkResourceExtension + implements BeforeAllCallback, AfterAllCallback, BeforeEachCallback, AfterEachCallback { private final FlinkResource delegate; + private boolean perClass; + private boolean anyTestFailed; + public FlinkResourceExtension(FlinkResource delegate) { this.delegate = delegate; } @@ -36,13 +45,32 @@ public class FlinkResourceExtension implements BeforeEachCallback, AfterEachCall } @Override - public void beforeEach(ExtensionContext context) throws Exception { + public void beforeAll(ExtensionContext context) throws Exception { + perClass = true; delegate.before(); } + @Override + public void afterAll(ExtensionContext context) { + if (anyTestFailed || context.getExecutionException().isPresent()) { + delegate.afterTestFailure(); + } else { + delegate.afterTestSuccess(); + } + } + + @Override + public void beforeEach(ExtensionContext context) throws Exception { + if (!perClass) { + delegate.before(); + } + } + @Override public void afterEach(ExtensionContext context) { - if (context.getExecutionException().isPresent()) { + if (perClass) { + anyTestFailed |= context.getExecutionException().isPresent(); + } else if (context.getExecutionException().isPresent()) { delegate.afterTestFailure(); } else { delegate.afterTestSuccess(); diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/CompileAndExecuteRemotePlanITCase.java b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/CompileAndExecuteRemotePlanITCase.java index 5ad87f0f87c..50c4f395eba 100644 --- a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/CompileAndExecuteRemotePlanITCase.java +++ b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/CompileAndExecuteRemotePlanITCase.java @@ -29,6 +29,7 @@ import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assumptions.assumeThat; @@ -43,7 +44,6 @@ class CompileAndExecuteRemotePlanITCase extends HdfsITCaseBase { @Override void createHDFS() { - super.createHDFS(); planDir = getRemotePlanDir(); } @@ -70,8 +70,8 @@ class CompileAndExecuteRemotePlanITCase extends HdfsITCaseBase { private String getRemotePlanDir() { return String.format( - "hdfs://%s:%s/foo/bar", - hdfsCluster.getURI().getHost(), hdfsCluster.getNameNodePort()); + "hdfs://%s:%s/foo/bar-%s", + hdfsCluster.getURI().getHost(), hdfsCluster.getNameNodePort(), UUID.randomUUID()); } private void assertPlanExists() throws IOException { diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/HdfsITCaseBase.java b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/HdfsITCaseBase.java index 0342c6bd517..b51649b71fe 100644 --- a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/HdfsITCaseBase.java +++ b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/HdfsITCaseBase.java @@ -25,11 +25,12 @@ import org.apache.flink.util.FileUtils; import org.apache.flink.util.OperatingSystem; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.fs.FileUtil; import org.apache.hadoop.hdfs.MiniDFSCluster; +import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.io.TempDir; import java.io.File; import java.io.FileNotFoundException; @@ -48,14 +49,32 @@ abstract class HdfsITCaseBase extends SqlITCaseBase { private static final Path HADOOP_CLASSPATH = ResourceTestUtils.getResource(".*hadoop.classpath"); - Configuration hdConf; - MiniDFSCluster hdfsCluster; + static Configuration hdConf; + static MiniDFSCluster hdfsCluster; + + @TempDir private static Path hdfsBaseDir; @BeforeAll - static void verifyOS() { + static void startHDFS() { assumeThat(OperatingSystem.isWindows()) .as("HDFS cluster cannot be started on Windows without extensions.") .isFalse(); + try { + hdConf = new Configuration(); + hdConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, hdfsBaseDir.toString()); + MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(hdConf); + hdfsCluster = builder.build(); + } catch (Throwable e) { + fail("Failed to create HDFS env" + e.getMessage(), e); + } + } + + @AfterAll + static void stopHDFS() { + if (hdfsCluster != null) { + hdfsCluster.shutdown(); + hdfsCluster = null; + } } @BeforeEach @@ -69,24 +88,11 @@ abstract class HdfsITCaseBase extends SqlITCaseBase { destroyHDFS(); } - void createHDFS() { - try { - hdConf = new Configuration(); - File baseDir = - new File(String.format("./target/hdfs/%s", getClass().getSimpleName())) - .getAbsoluteFile(); - FileUtil.fullyDelete(baseDir); - hdConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, baseDir.getAbsolutePath()); - MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(hdConf); - hdfsCluster = builder.build(); - } catch (Throwable e) { - fail("Failed to create HDFS env" + e.getMessage(), e); - } - } + /** Per-test setup on the HDFS cluster shared by the class. */ + void createHDFS() {} - void destroyHDFS() { - hdfsCluster.shutdown(); - } + /** Per-test cleanup on the HDFS cluster shared by the class. */ + void destroyHDFS() {} @Override void executeSqlStatements( diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/SqlITCaseBase.java b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/SqlITCaseBase.java index 31ac671344b..559da69bdea 100644 --- a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/SqlITCaseBase.java +++ b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/SqlITCaseBase.java @@ -41,6 +41,8 @@ import org.apache.flink.types.RowKind; import org.apache.flink.util.Collector; import org.apache.flink.util.TestLoggerExtension; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.api.extension.RegisterExtension; @@ -81,7 +83,7 @@ abstract class SqlITCaseBase { } @RegisterExtension - private final FlinkResourceExtension flinkExtension = + private static final FlinkResourceExtension flinkExtension = new FlinkResourceExtension( new LocalStandaloneFlinkResourceFactory() .create( @@ -104,6 +106,21 @@ abstract class SqlITCaseBase { return flinkConfig; } + private static ClusterController clusterController; + + @BeforeAll + static void startCluster() throws Exception { + clusterController = flinkExtension.getFlinkResource().startCluster(1); + } + + @AfterAll + static void stopCluster() throws Exception { + if (clusterController != null) { + clusterController.close(); + clusterController = null; + } + } + @BeforeEach void before() throws Exception { LOG.info("The current temporary path: {}", tmp); @@ -136,22 +153,19 @@ abstract class SqlITCaseBase { Map<Path, Function<List<String>, List<String>>> formatters, List<URI> dependencies) throws Exception { - try (ClusterController clusterController = - flinkExtension.getFlinkResource().startCluster(1)) { - List<String> sqlLines = initializeSqlLines(sqlPath); - - executeSqlStatements(clusterController, sqlLines, dependencies); - - // Wait until all the results flushed to the json file. - LOG.info("Verify the result."); - for (Map.Entry<Path, List<String>> entry : resultItems.entrySet()) { - checkResultFile( - entry.getKey(), - entry.getValue(), - formatters.getOrDefault(entry.getKey(), this::formatRawResult)); - } - LOG.info("The SQL client test run successfully."); + List<String> sqlLines = initializeSqlLines(sqlPath); + + executeSqlStatements(clusterController, sqlLines, dependencies); + + // Wait until all the results flushed to the json file. + LOG.info("Verify the result."); + for (Map.Entry<Path, List<String>> entry : resultItems.entrySet()) { + checkResultFile( + entry.getKey(), + entry.getValue(), + formatters.getOrDefault(entry.getKey(), this::formatRawResult)); } + LOG.info("The SQL client test run successfully."); } Map<String, String> generateReplaceVars() { diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/UsingRemoteJarITCase.java b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/UsingRemoteJarITCase.java index 6a3ae1f8130..5477710bd6f 100644 --- a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/UsingRemoteJarITCase.java +++ b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/UsingRemoteJarITCase.java @@ -36,6 +36,7 @@ import java.net.URI; import java.util.Arrays; import java.util.Collections; import java.util.Map; +import java.util.UUID; import static org.assertj.core.api.Assertions.fail; @@ -68,8 +69,7 @@ class UsingRemoteJarITCase extends HdfsITCaseBase { @Override void createHDFS() { - super.createHDFS(); - hdPath = new org.apache.hadoop.fs.Path("/test.jar"); + hdPath = new org.apache.hadoop.fs.Path("/" + UUID.randomUUID() + "/test.jar"); try { hdfs = hdPath.getFileSystem(hdConf); @@ -83,11 +83,10 @@ class UsingRemoteJarITCase extends HdfsITCaseBase { @Override void destroyHDFS() { try { - hdfs.delete(hdPath, false); + hdfs.delete(hdPath.getParent(), true); } catch (IOException e) { fail("Failed to cleanup HDFS path" + e.getMessage(), e); } - super.destroyHDFS(); } @TestTemplate
