[HUDI-2785] Make HoodieParquetWriter Thread safe and memory executor …

…graceful exit

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieParquetWriter.java

@@ -74,7 +74,7 @@ public HoodieParquetWriter(String instantTime, Path file, HoodieAvroParquetConfi
   }
 
   @Override
-  public void writeAvroWithMetadata(R avroRecord, HoodieRecord record) throws IOException {
+  public synchronized void writeAvroWithMetadata(R avroRecord, HoodieRecord record) throws IOException {
     if (populateMetaFields) {
       prepRecordWithMetadata(avroRecord, record, instantTime,
           taskContextSupplier.getPartitionIdSupplier().get(), recordIndex, file.getName());
@@ -91,13 +91,18 @@ public boolean canWrite() {
   }
 
   @Override
-  public void writeAvro(String key, IndexedRecord object) throws IOException {
+  public synchronized void writeAvro(String key, IndexedRecord object) throws IOException {
     super.write(object);
     if (populateMetaFields) {
       writeSupport.add(key);
     }
   }
 
+  @Override
+  public synchronized void close() throws IOException {
+    super.close();
+  }
+
   @Override
   public long getBytesWritten() {
     return fs.getBytesWritten(file);

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkMergeHelper.java

@@ -101,13 +101,13 @@ public void runMerge(HoodieTable<T, JavaRDD<HoodieRecord<T>>, JavaRDD<HoodieKey>
     } catch (Exception e) {
       throw new HoodieException(e);
     } finally {
+      if (null != wrapper) {
+        wrapper.shutdownNow();
+      }
       if (reader != null) {
         reader.close();
       }
       mergeHandle.close();
-      if (null != wrapper) {
-        wrapper.shutdownNow();
-      }
     }
   }
 }

hudi-common/src/main/java/org/apache/hudi/common/util/queue/BoundedInMemoryExecutor.java

@@ -34,6 +34,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -48,6 +49,7 @@ public class BoundedInMemoryExecutor<I, O, E> {
 
   // Executor service used for launching writer thread.
   private final ExecutorService executorService;
+  public static final int TERMINATE_WAIT_TIME = 60;
   // Used for buffering records which is controlled by HoodieWriteConfig#WRITE_BUFFER_LIMIT_BYTES.
   private final BoundedInMemoryQueue<I, O> queue;
   // Producers
@@ -154,6 +156,11 @@ public boolean isRemaining() {
 
   public void shutdownNow() {
     executorService.shutdownNow();
+    try {
+      executorService.awaitTermination(TERMINATE_WAIT_TIME, TimeUnit.SECONDS);
+    } catch (InterruptedException e) {
+      LOG.error(String.format("executor didn't terminate within %d seconds", TERMINATE_WAIT_TIME), e);
+    }
   }
 
   public BoundedInMemoryQueue<I, O> getQueue() {

hudi-common/src/main/java/org/apache/hudi/common/util/queue/BoundedInMemoryQueue.java

@@ -179,7 +179,7 @@ public void insertRecord(I t) throws Exception {
     if (isWriteDone.get()) {
       throw new IllegalStateException("Queue closed for enqueueing new entries");
     }
-
+    checkIfInterrupted();
     // We need to stop queueing if queue-reader has failed and exited.
     throwExceptionIfFailed();
 
@@ -204,6 +204,7 @@ private boolean expectMoreRecords() {
    * singleton iterator for this queue.
    */
   private Option<O> readNextRecord() {
+    checkIfInterrupted();
     if (this.isReadDone.get()) {
       return Option.empty();
     }
@@ -249,6 +250,13 @@ private void throwExceptionIfFailed() {
     }
   }
 
+  private void checkIfInterrupted() {
+    if (Thread.currentThread().isInterrupted()) {
+      this.hasFailed.set(new InterruptedException());
+      Thread.currentThread().interrupt();
+    }
+  }
+
   /**
    * API to allow producers and consumer to communicate termination due to failure.
    */