Revisited HoodieMergeHelper shutdown seq to defer to the Executor to appropriatly clean up the resources

Author: Alexey Kudinkin

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseMergeHelper.java

@@ -18,17 +18,19 @@
 
 package org.apache.hudi.table.action.commit;
 
-import org.apache.avro.generic.GenericRecord;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
 import org.apache.hudi.avro.HoodieAvroUtils;
-import org.apache.hudi.client.utils.MergingIterator;
+import org.apache.hudi.client.utils.ClosableMergingIterator;
+import org.apache.hudi.common.util.ClosableIterator;
 import org.apache.hudi.common.util.queue.HoodieConsumer;
 import org.apache.hudi.io.HoodieMergeHandle;
 import org.apache.hudi.io.storage.HoodieFileReader;
 import org.apache.hudi.io.storage.HoodieFileReaderFactory;
 import org.apache.hudi.table.HoodieTable;
 
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+
 import java.io.IOException;
 import java.util.Iterator;
 
@@ -51,14 +53,14 @@ public abstract class BaseMergeHelper {
    * for indexing, writing and other functionality.
    *
    */
-  protected Iterator<GenericRecord> getMergingIterator(HoodieTable<?, ?, ?, ?> table,
-                                                       HoodieMergeHandle<?, ?, ?, ?> mergeHandle,
-                                                       Path bootstrapFilePath,
-                                                       Iterator<GenericRecord> recordIterator) throws IOException {
+  protected ClosableIterator<GenericRecord> getMergingIterator(HoodieTable<?, ?, ?, ?> table,
+                                                               HoodieMergeHandle<?, ?, ?, ?> mergeHandle,
+                                                               Path bootstrapFilePath,
+                                                               Iterator<GenericRecord> recordIterator) throws IOException {
     Configuration bootstrapFileConfig = new Configuration(table.getHadoopConf());
     HoodieFileReader<GenericRecord> bootstrapReader =
         HoodieFileReaderFactory.getFileReader(bootstrapFileConfig, bootstrapFilePath);
-    return new MergingIterator<>(recordIterator, bootstrapReader.getRecordIterator(),
+    return new ClosableMergingIterator<>(recordIterator, bootstrapReader.getRecordIterator(),
         (inputRecordPair) -> HoodieAvroUtils.stitchRecords(inputRecordPair.getLeft(), inputRecordPair.getRight(), mergeHandle.getWriterSchemaWithMetaFields()));
   }
 

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java

@@ -31,7 +31,7 @@
 import org.apache.hudi.common.util.ClosableIterator;
 import org.apache.hudi.common.util.InternalSchemaCache;
 import org.apache.hudi.common.util.Option;
-import org.apache.hudi.common.util.collection.MappingIterator;
+import org.apache.hudi.common.util.collection.ClosableMappingIterator;
 import org.apache.hudi.common.util.queue.HoodieExecutor;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieException;
@@ -48,7 +48,6 @@
 import org.apache.hudi.util.QueueBasedExecutorFactory;
 
 import java.io.IOException;
-import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -102,7 +101,7 @@ public void runMerge(HoodieTable<?, ?, ?, ?> table,
     HoodieExecutor<GenericRecord, GenericRecord, Void> wrapper = null;
 
     try {
-      Iterator<GenericRecord> recordIterator;
+      ClosableIterator<GenericRecord> recordIterator;
 
       // In case writer's schema is simply a projection of the reader's one we can read
       // the records in the projected schema directly
@@ -112,7 +111,7 @@ public void runMerge(HoodieTable<?, ?, ?, ?> table,
         Path bootstrapFilePath = new Path(baseFile.getBootstrapBaseFile().get().getPath());
         recordIterator = getMergingIterator(table, mergeHandle, bootstrapFilePath, baseFileRecordIterator);
       } else if (schemaEvolutionTransformerOpt.isPresent()) {
-        recordIterator = new MappingIterator<>(baseFileRecordIterator,
+        recordIterator = new ClosableMappingIterator<>(baseFileRecordIterator,
             schemaEvolutionTransformerOpt.get());
       } else {
         recordIterator = baseFileRecordIterator;
@@ -130,14 +129,10 @@ public void runMerge(HoodieTable<?, ?, ?, ?> table,
     } catch (Exception e) {
       throw new HoodieException(e);
     } finally {
-      // HUDI-2875: mergeHandle is not thread safe, we should totally terminate record inputting
-      // and executor firstly and then close mergeHandle.
-      reader.close();
-      if (null != wrapper) {
+      if (wrapper != null) {
         wrapper.shutdownNow();
         wrapper.awaitTermination();
       }
-      mergeHandle.close();
     }
   }