[MINOR] Fix zookeeper session expiration bug (#10671)

apache · Feb 27, 2024 · 5b94afa · 5b94afa
1 parent cad5605
commit 5b94afa
Show file tree

Hide file tree

Showing 19 changed files with 129 additions and 74 deletions.
diff --git a/...st/src/test/java/org/apache/hudi/integ/testsuite/TestDFSHoodieTestSuiteWriterAdapter.java b/...st/src/test/java/org/apache/hudi/integ/testsuite/TestDFSHoodieTestSuiteWriterAdapter.java
@@ -69,7 +69,7 @@ public static void initClass() throws Exception {
   }
 
   @AfterAll
-  public static void cleanupClass() {
+  public static void cleanupClass() throws IOException {
     UtilitiesTestBase.cleanUpUtilitiesTestServices();
   }
 

diff --git a/hudi-integ-test/src/test/java/org/apache/hudi/integ/testsuite/TestFileDeltaInputWriter.java b/hudi-integ-test/src/test/java/org/apache/hudi/integ/testsuite/TestFileDeltaInputWriter.java
@@ -63,7 +63,7 @@ public static void initClass() throws Exception {
   }
 
   @AfterAll
-  public static void cleanupClass() {
+  public static void cleanupClass() throws IOException {
     UtilitiesTestBase.cleanUpUtilitiesTestServices();
   }
 

diff --git a/...-integ-test/src/test/java/org/apache/hudi/integ/testsuite/job/TestHoodieTestSuiteJob.java b/...-integ-test/src/test/java/org/apache/hudi/integ/testsuite/job/TestHoodieTestSuiteJob.java
@@ -49,6 +49,7 @@
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 
+import java.io.IOException;
 import java.util.UUID;
 import java.util.stream.Stream;
 
@@ -134,7 +135,7 @@ public static void initClass() throws Exception {
   }
 
   @AfterAll
-  public static void cleanupClass() {
+  public static void cleanupClass() throws IOException {
     UtilitiesTestBase.cleanUpUtilitiesTestServices();
   }
 

diff --git a/...est/src/test/java/org/apache/hudi/integ/testsuite/reader/TestDFSAvroDeltaInputReader.java b/...est/src/test/java/org/apache/hudi/integ/testsuite/reader/TestDFSAvroDeltaInputReader.java
@@ -48,7 +48,7 @@ public static void initClass() throws Exception {
   }
 
   @AfterAll
-  public static void cleanupClass() {
+  public static void cleanupClass() throws IOException {
     UtilitiesTestBase.cleanUpUtilitiesTestServices();
   }
 

diff --git a/...src/test/java/org/apache/hudi/integ/testsuite/reader/TestDFSHoodieDatasetInputReader.java b/...src/test/java/org/apache/hudi/integ/testsuite/reader/TestDFSHoodieDatasetInputReader.java
@@ -38,6 +38,7 @@
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
+import java.io.IOException;
 import java.util.HashSet;
 import java.util.List;
 
@@ -55,7 +56,7 @@ public static void initClass() throws Exception {
   }
 
   @AfterAll
-  public static void cleanupClass() {
+  public static void cleanupClass() throws IOException {
     UtilitiesTestBase.cleanUpUtilitiesTestServices();
   }
 

diff --git a/...utilities/src/test/java/org/apache/hudi/utilities/callback/TestKafkaCallbackProvider.java b/...utilities/src/test/java/org/apache/hudi/utilities/callback/TestKafkaCallbackProvider.java
@@ -30,9 +30,12 @@
 import org.apache.hudi.utilities.testutils.UtilitiesTestBase;
 import org.apache.spark.streaming.kafka010.KafkaTestUtils;
 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.Test;
 
+import java.io.IOException;
 import java.util.List;
 import java.util.UUID;
 
@@ -43,19 +46,27 @@
 public class TestKafkaCallbackProvider extends UtilitiesTestBase {
   private final String testTopicName = "hoodie_test_" + UUID.randomUUID();
 
-  private static KafkaTestUtils testUtils;
+  private KafkaTestUtils testUtils;
 
   @BeforeAll
   public static void initClass() throws Exception {
     UtilitiesTestBase.initTestServices();
+  }
+
+  @BeforeEach
+  public void setup() {
     testUtils = new KafkaTestUtils();
     testUtils.setup();
   }
 
+  @AfterEach
+  public void tearDown() {
+    testUtils.teardown();
+  }
+
   @AfterAll
-  public static void cleanupClass() {
+  public static void cleanupClass() throws IOException {
     UtilitiesTestBase.cleanUpUtilitiesTestServices();
-    testUtils.teardown();
   }
 
   @Test

diff --git a/...es/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java b/...es/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java
@@ -51,6 +51,7 @@
 import org.apache.spark.sql.SQLContext;
 import org.apache.spark.streaming.kafka010.KafkaTestUtils;
 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.slf4j.Logger;
@@ -130,14 +131,15 @@ public class HoodieDeltaStreamerTestBase extends UtilitiesTestBase {
   static final String HOODIE_CONF_PARAM = "--hoodie-conf";
   static final String HOODIE_CONF_VALUE1 = "hoodie.datasource.hive_sync.table=test_table";
   static final String HOODIE_CONF_VALUE2 = "hoodie.datasource.write.recordkey.field=Field1,Field2,Field3";
-  public static KafkaTestUtils testUtils;
   protected static String topicName;
   protected static String defaultSchemaProviderClassName = FilebasedSchemaProvider.class.getName();
   protected static int testNum = 1;
 
   Map<String, String> hudiOpts = new HashMap<>();
+  public KafkaTestUtils testUtils;
 
-  protected static void prepareTestSetup() throws IOException {
+  @BeforeEach
+  protected void prepareTestSetup() throws IOException {
     PARQUET_SOURCE_ROOT = basePath + "/parquetFiles";
     ORC_SOURCE_ROOT = basePath + "/orcFiles";
     JSON_KAFKA_SOURCE_ROOT = basePath + "/jsonKafkaFiles";
@@ -245,16 +247,15 @@ protected static void writeCommonPropsToFile(FileSystem dfs, String dfsBasePath)
   @BeforeAll
   public static void initClass() throws Exception {
     UtilitiesTestBase.initTestServices(false, true, false);
-    prepareTestSetup();
   }
 
   @AfterAll
-  public static void tearDown() {
-    cleanupKafkaTestUtils();
+  public static void tearDown() throws IOException {
     UtilitiesTestBase.cleanUpUtilitiesTestServices();
   }
 
-  public static void cleanupKafkaTestUtils() {
+  @AfterEach
+  public void cleanupKafkaTestUtils() {
     if (testUtils != null) {
       testUtils.teardown();
     }

diff --git a/...lities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java b/...lities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
@@ -1716,11 +1716,11 @@ public void testDistributedTestDataSource() {
     assertEquals(1000, c);
   }
 
-  private static void prepareJsonKafkaDFSFiles(int numRecords, boolean createTopic, String topicName) {
+  private void prepareJsonKafkaDFSFiles(int numRecords, boolean createTopic, String topicName) {
     prepareJsonKafkaDFSFiles(numRecords, createTopic, topicName, 2);
   }
 
-  private static void prepareJsonKafkaDFSFiles(int numRecords, boolean createTopic, String topicName, int numPartitions) {
+  private void prepareJsonKafkaDFSFiles(int numRecords, boolean createTopic, String topicName, int numPartitions) {
     if (createTopic) {
       try {
         testUtils.createTopic(topicName, numPartitions);

diff --git a/...a/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamerSchemaEvolutionBase.java b/...a/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamerSchemaEvolutionBase.java
@@ -129,7 +129,6 @@ public void teardown() throws Exception {
   @AfterAll
   static void teardownAll() {
     defaultSchemaProviderClassName = FilebasedSchemaProvider.class.getName();
-    HoodieDeltaStreamerTestBase.cleanupKafkaTestUtils();
   }
 
   protected HoodieStreamer deltaStreamer;

diff --git a/...utilities/src/test/java/org/apache/hudi/utilities/schema/TestFilebasedSchemaProvider.java b/...utilities/src/test/java/org/apache/hudi/utilities/schema/TestFilebasedSchemaProvider.java
@@ -51,7 +51,7 @@ public static void initClass() throws Exception {
   }
 
   @AfterAll
-  public static void cleanUpUtilitiesTestServices() {
+  public static void cleanUpUtilitiesTestServices() throws IOException {
     UtilitiesTestBase.cleanUpUtilitiesTestServices();
   }
 

diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/BaseTestKafkaSource.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/BaseTestKafkaSource.java
@@ -38,8 +38,8 @@
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
 import org.apache.spark.streaming.kafka010.KafkaTestUtils;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import java.util.ArrayList;
@@ -58,20 +58,20 @@
  */
 abstract class BaseTestKafkaSource extends SparkClientFunctionalTestHarness {
   protected static final String TEST_TOPIC_PREFIX = "hoodie_test_";
-  protected static KafkaTestUtils testUtils;
 
   protected final HoodieIngestionMetrics metrics = mock(HoodieIngestionMetrics.class);
 
   protected SchemaProvider schemaProvider;
+  protected KafkaTestUtils testUtils;
 
-  @BeforeAll
-  public static void initClass() {
+  @BeforeEach
+  public void initClass() {
     testUtils = new KafkaTestUtils();
     testUtils.setup();
   }
 
-  @AfterAll
-  public static void cleanupClass() {
+  @AfterEach
+  public void cleanupClass() {
     testUtils.teardown();
   }
 

diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java
@@ -45,8 +45,9 @@
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
 import org.apache.spark.streaming.kafka010.KafkaTestUtils;
-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.Test;
 
 import java.io.IOException;
@@ -68,8 +69,6 @@
 public class TestAvroKafkaSource extends SparkClientFunctionalTestHarness {
   protected static final String TEST_TOPIC_PREFIX = "hoodie_avro_test_";
 
-  protected static KafkaTestUtils testUtils;
-
   protected static HoodieTestDataGenerator dataGen;
 
   protected static String SCHEMA_PATH = "/tmp/schema_file.avsc";
@@ -78,15 +77,21 @@ public class TestAvroKafkaSource extends SparkClientFunctionalTestHarness {
 
   protected SchemaProvider schemaProvider;
 
+  protected KafkaTestUtils testUtils;
+
   @BeforeAll
   public static void initClass() {
-    testUtils = new KafkaTestUtils();
     dataGen = new HoodieTestDataGenerator(0xDEED);
+  }
+
+  @BeforeEach
+  public void setup() {
+    testUtils = new KafkaTestUtils();
     testUtils.setup();
   }
 
-  @AfterAll
-  public static void cleanupClass() {
+  @AfterEach
+  public void tearDown() {
     testUtils.teardown();
   }
 

diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestSqlFileBasedSource.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestSqlFileBasedSource.java
@@ -28,7 +28,6 @@
 import org.apache.hudi.utilities.testutils.UtilitiesTestBase;
 
 import org.apache.avro.generic.GenericRecord;
-import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.spark.api.java.JavaRDD;
 import org.apache.spark.sql.AnalysisException;
@@ -64,17 +63,10 @@ public class TestSqlFileBasedSource extends UtilitiesTestBase {
   @BeforeAll
   public static void initClass() throws Exception {
     UtilitiesTestBase.initTestServices(false, true, false);
-    FileSystem fs = UtilitiesTestBase.fs;
-    UtilitiesTestBase.Helpers.copyToDFS(
-        "streamer-config/sql-file-based-source.sql", fs,
-        UtilitiesTestBase.basePath + "/sql-file-based-source.sql");
-    UtilitiesTestBase.Helpers.copyToDFS(
-        "streamer-config/sql-file-based-source-invalid-table.sql", fs,
-        UtilitiesTestBase.basePath + "/sql-file-based-source-invalid-table.sql");
   }
 
   @AfterAll
-  public static void cleanupClass() {
+  public static void cleanupClass() throws IOException {
     UtilitiesTestBase.cleanUpUtilitiesTestServices();
   }
 
@@ -113,7 +105,11 @@ private void generateTestTable(String filename, String instantTime, int n) throw
    * @throws IOException
    */
   @Test
-  public void testSqlFileBasedSourceAvroFormat() {
+  public void testSqlFileBasedSourceAvroFormat() throws IOException {
+    UtilitiesTestBase.Helpers.copyToDFS(
+        "streamer-config/sql-file-based-source.sql", fs,
+        UtilitiesTestBase.basePath + "/sql-file-based-source.sql");
+
     props.setProperty(sqlFileSourceConfig, UtilitiesTestBase.basePath + "/sql-file-based-source.sql");
     sqlFileSource = new SqlFileBasedSource(props, jsc, sparkSession, schemaProvider);
     sourceFormatAdapter = new SourceFormatAdapter(sqlFileSource);
@@ -136,7 +132,11 @@ public void testSqlFileBasedSourceAvroFormat() {
    * @throws IOException
    */
   @Test
-  public void testSqlFileBasedSourceRowFormat() {
+  public void testSqlFileBasedSourceRowFormat() throws IOException {
+    UtilitiesTestBase.Helpers.copyToDFS(
+        "streamer-config/sql-file-based-source.sql", fs,
+        UtilitiesTestBase.basePath + "/sql-file-based-source.sql");
+
     props.setProperty(sqlFileSourceConfig, UtilitiesTestBase.basePath + "/sql-file-based-source.sql");
     sqlFileSource = new SqlFileBasedSource(props, jsc, sparkSession, schemaProvider);
     sourceFormatAdapter = new SourceFormatAdapter(sqlFileSource);
@@ -154,7 +154,11 @@ public void testSqlFileBasedSourceRowFormat() {
    * @throws IOException
    */
   @Test
-  public void testSqlFileBasedSourceMoreRecordsThanSourceLimit() {
+  public void testSqlFileBasedSourceMoreRecordsThanSourceLimit() throws IOException {
+    UtilitiesTestBase.Helpers.copyToDFS(
+        "streamer-config/sql-file-based-source.sql", fs,
+        UtilitiesTestBase.basePath + "/sql-file-based-source.sql");
+
     props.setProperty(sqlFileSourceConfig, UtilitiesTestBase.basePath + "/sql-file-based-source.sql");
     sqlFileSource = new SqlFileBasedSource(props, jsc, sparkSession, schemaProvider);
     sourceFormatAdapter = new SourceFormatAdapter(sqlFileSource);
@@ -171,7 +175,11 @@ public void testSqlFileBasedSourceMoreRecordsThanSourceLimit() {
    * @throws IOException
    */
   @Test
-  public void testSqlFileBasedSourceInvalidTable() {
+  public void testSqlFileBasedSourceInvalidTable() throws IOException {
+    UtilitiesTestBase.Helpers.copyToDFS(
+        "streamer-config/sql-file-based-source-invalid-table.sql", fs,
+        UtilitiesTestBase.basePath + "/sql-file-based-source-invalid-table.sql");
+
     props.setProperty(sqlFileSourceConfig, UtilitiesTestBase.basePath + "/sql-file-based-source-invalid-table.sql");
     sqlFileSource = new SqlFileBasedSource(props, jsc, sparkSession, schemaProvider);
     sourceFormatAdapter = new SourceFormatAdapter(sqlFileSource);
@@ -182,7 +190,11 @@ public void testSqlFileBasedSourceInvalidTable() {
   }
 
   @Test
-  public void shouldSetCheckpointForSqlFileBasedSourceWithEpochCheckpoint() {
+  public void shouldSetCheckpointForSqlFileBasedSourceWithEpochCheckpoint() throws IOException {
+    UtilitiesTestBase.Helpers.copyToDFS(
+        "streamer-config/sql-file-based-source.sql", fs,
+        UtilitiesTestBase.basePath + "/sql-file-based-source.sql");
+
     props.setProperty(sqlFileSourceConfig, UtilitiesTestBase.basePath + "/sql-file-based-source.sql");
     props.setProperty(sqlFileSourceConfigEmitChkPointConf, "true");
 

diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestSqlSource.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestSqlSource.java
@@ -64,7 +64,7 @@ public static void initClass() throws Exception {
   }
 
   @AfterAll
-  public static void cleanupClass() {
+  public static void cleanupClass() throws IOException {
     UtilitiesTestBase.cleanUpUtilitiesTestServices();
   }