[FLINK-24077][HBase/IT] use MiniClusterWithClientResource as @ClassRule.

while using TableEnvironment in the ITCase, a Flink MiniCluster will be started/stopped automatically in the background. Since the shutdown of the MiniCluster will be called asynchronously, CollectResultFetcher will get data lost sometimes based on race conditions and the unchecked RuntimeException java.lang.IllegalStateException will be thrown that we were not aware of.

The solution is to control the lifecycle of the MiniCluster manually in this test. The MiniClusterWithClientResource could be a good fit in this case.

(cherry picked from commit fca04c3)

Author: JingGe

flink-connectors/flink-connector-hbase-2.2/src/test/java/org/apache/flink/connector/hbase2/HBaseConnectorITCase.java

@@ -25,6 +25,7 @@
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.tuple.Tuple1;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.connector.hbase.util.HBaseTableSchema;
 import org.apache.flink.connector.hbase.util.PlannerType;
 import org.apache.flink.connector.hbase2.source.AbstractTableInputFormat;
@@ -33,6 +34,7 @@
 import org.apache.flink.connector.hbase2.source.HBaseRowInputFormat;
 import org.apache.flink.connector.hbase2.source.HBaseTableSource;
 import org.apache.flink.connector.hbase2.util.HBaseTestBase;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.api.DataTypes;
@@ -46,6 +48,7 @@
 import org.apache.flink.table.descriptors.HBase;
 import org.apache.flink.table.descriptors.Schema;
 import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
 import org.apache.flink.test.util.TestBaseUtils;
 import org.apache.flink.types.Row;
 import org.apache.flink.types.RowKind;
@@ -57,6 +60,7 @@
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
@@ -77,6 +81,13 @@
 @RunWith(Parameterized.class)
 public class HBaseConnectorITCase extends HBaseTestBase {
 
+    @ClassRule
+    public static final MiniClusterWithClientResource MINI_CLUSTER =
+            new MiniClusterWithClientResource(
+                    new MiniClusterResourceConfiguration.Builder()
+                            .setConfiguration(new Configuration())
+                            .build());
+
     @Parameterized.Parameter public PlannerType planner;
 
     @Parameterized.Parameter(1)
@@ -436,8 +447,6 @@ public void testTableSink() throws Exception {
                                 + " AS h");
 
         TableResult tableResult2 = table.execute();
-        // wait to finish
-        tableResult2.getJobClient().get().getJobExecutionResult().get();
 
         List<Row> results = CollectionUtil.iteratorToList(tableResult2.collect());
 
@@ -529,8 +538,6 @@ public void testTableSourceSinkWithDDL() throws Exception {
                         + " AS h";
 
         TableResult tableResult3 = batchEnv.executeSql(query);
-        // wait to finish
-        tableResult3.getJobClient().get().getJobExecutionResult().get();
 
         List<String> result =
                 Lists.newArrayList(tableResult3.collect()).stream()