[MINOR] Performance improvement of flink ITs with reused miniCluster

Author: trushev

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/ITTestDataStreamWrite.java

@@ -29,6 +29,7 @@
 import org.apache.hudi.util.AvroSchemaConverter;
 import org.apache.hudi.util.HoodiePipeline;
 import org.apache.hudi.util.StreamerUtil;
+import org.apache.hudi.utils.AbstractHoodieTestBase;
 import org.apache.hudi.utils.TestConfigurations;
 import org.apache.hudi.utils.TestData;
 import org.apache.hudi.utils.TestUtils;
@@ -52,7 +53,6 @@
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
 import org.apache.flink.table.types.logical.RowType;
-import org.apache.flink.util.TestLogger;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -71,7 +71,7 @@
 /**
  * Integration test for Flink Hoodie stream sink.
  */
-public class ITTestDataStreamWrite extends TestLogger {
+public class ITTestDataStreamWrite extends AbstractHoodieTestBase {
 
   private static final Map<String, List<String>> EXPECTED = new HashMap<>();
   private static final Map<String, List<String>> EXPECTED_TRANSFORMER = new HashMap<>();

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestHoodieFlinkClustering.java

@@ -39,6 +39,7 @@
 import org.apache.hudi.util.AvroSchemaConverter;
 import org.apache.hudi.util.CompactionUtil;
 import org.apache.hudi.util.StreamerUtil;
+import org.apache.hudi.utils.AbstractHoodieTestBase;
 import org.apache.hudi.utils.TestConfigurations;
 import org.apache.hudi.utils.TestData;
 import org.apache.hudi.utils.TestSQL;
@@ -69,7 +70,7 @@
 /**
  * IT cases for {@link HoodieFlinkClusteringJob}.
  */
-public class ITTestHoodieFlinkClustering {
+public class ITTestHoodieFlinkClustering extends AbstractHoodieTestBase {
 
   private static final Map<String, String> EXPECTED = new HashMap<>();
 

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/ITTestHoodieFlinkCompactor.java

@@ -30,6 +30,7 @@
 import org.apache.hudi.table.HoodieFlinkTable;
 import org.apache.hudi.util.CompactionUtil;
 import org.apache.hudi.util.StreamerUtil;
+import org.apache.hudi.utils.AbstractHoodieTestBase;
 import org.apache.hudi.utils.TestConfigurations;
 import org.apache.hudi.utils.TestData;
 import org.apache.hudi.utils.TestSQL;
@@ -63,7 +64,7 @@
 /**
  * IT cases for {@link org.apache.hudi.common.model.HoodieRecord}.
  */
-public class ITTestHoodieFlinkCompactor {
+public class ITTestHoodieFlinkCompactor extends AbstractHoodieTestBase {
 
   protected static final Logger LOG = LoggerFactory.getLogger(ITTestHoodieFlinkCompactor.class);
 
@@ -155,7 +156,7 @@ public void testHoodieFlinkCompactor(boolean enableChangelog) throws Exception {
         .transform("compact_task",
             TypeInformation.of(CompactionCommitEvent.class),
             new ProcessOperator<>(new CompactFunction(conf)))
-        .setParallelism(compactionPlan.getOperations().size())
+        .setParallelism(4)
         .addSink(new CompactionCommitSink(conf))
         .name("clean_commits")
         .uid("uid_clean_commits")
@@ -195,6 +196,7 @@ public void testHoodieFlinkCompactorService(boolean enableChangelog) throws Exce
     cfg.schedule = true;
     Configuration conf = FlinkCompactionConfig.toFlinkConfig(cfg);
     conf.setString(FlinkOptions.TABLE_TYPE.key(), "MERGE_ON_READ");
+    conf.setInteger(FlinkOptions.COMPACTION_TASKS.key(), 4);
 
     HoodieFlinkCompactor.AsyncCompactionService asyncCompactionService = new HoodieFlinkCompactor.AsyncCompactionService(cfg, conf, env);
     asyncCompactionService.start(null);

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java

@@ -27,6 +27,7 @@
 import org.apache.hudi.table.catalog.HoodieCatalogTestUtils;
 import org.apache.hudi.table.catalog.HoodieHiveCatalog;
 import org.apache.hudi.util.StreamerUtil;
+import org.apache.hudi.utils.AbstractHoodieTestBase;
 import org.apache.hudi.utils.TestConfigurations;
 import org.apache.hudi.utils.TestData;
 import org.apache.hudi.utils.TestSQL;
@@ -44,7 +45,6 @@
 import org.apache.flink.table.catalog.ObjectPath;
 import org.apache.flink.table.catalog.exceptions.TableNotExistException;
 import org.apache.flink.table.data.RowData;
-import org.apache.flink.test.util.AbstractTestBase;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.CollectionUtil;
 import org.junit.jupiter.api.BeforeEach;
@@ -77,7 +77,7 @@
 /**
  * IT cases for Hoodie table source and sink.
  */
-public class ITTestHoodieDataSource extends AbstractTestBase {
+public class ITTestHoodieDataSource extends AbstractHoodieTestBase {
   private TableEnvironment streamTableEnv;
   private TableEnvironment batchTableEnv;
 

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/AbstractHoodieTestBase.java

@@ -0,0 +1,76 @@
+/*
+ * 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.hudi.utils;
+
+import org.apache.hudi.exception.HoodieException;
+
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+
+import java.lang.reflect.Field;
+import java.util.Arrays;
+
+import static java.lang.reflect.Modifier.isPublic;
+import static java.lang.reflect.Modifier.isStatic;
+
+/**
+ * Hoodie base class for tests that run multiple tests and want to reuse the same Flink cluster.
+ * Unlike {@link AbstractTestBase}, this class is designed to run with JUnit 5.
+ */
+public abstract class AbstractHoodieTestBase extends AbstractTestBase {
+
+  private static final MiniClusterWithClientResource MINI_CLUSTER_RESOURCE = getMiniClusterFromParentClass();
+
+  @BeforeAll
+  public static void beforeAll() throws Exception {
+    MINI_CLUSTER_RESOURCE.before();
+  }
+
+  @AfterAll
+  public static void afterAll() {
+    MINI_CLUSTER_RESOURCE.after();
+  }
+
+  @AfterEach
+  public void afterEach() throws Exception {
+    cleanupRunningJobs();
+  }
+
+  private static MiniClusterWithClientResource getMiniClusterFromParentClass() {
+    String fieldFlink114 = "miniClusterResource";
+    String fieldFlink115 = "MINI_CLUSTER_RESOURCE";
+    Field miniClusterField = Arrays.stream(AbstractTestBase.class.getDeclaredFields())
+        .filter(f -> isPublic(f.getModifiers()) && isStatic(f.getModifiers()))
+        .filter(f -> fieldFlink114.equals(f.getName()) || fieldFlink115.equals(f.getName()))
+        .findAny()
+        .orElseThrow(() -> new NoSuchFieldError(String.format(
+            "%s not found in %s",
+            fieldFlink115,
+            AbstractTestBase.class.getName())));
+    try {
+      return (MiniClusterWithClientResource) miniClusterField.get(null);
+    } catch (IllegalAccessException e) {
+      throw new HoodieException("Cannot access miniCluster field", e);
+    }
+  }
+}