address comments

Author: yuezhang

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/util/ExecutorFactory.java

@@ -55,7 +55,7 @@ public static <I, O, E> HoodieExecutor<E> create(HoodieWriteConfig hoodieConfig,
         return new DisruptorExecutor<>(hoodieConfig.getDisruptorWriteBufferSize(), inputItr, consumer,
             transformFunction, hoodieConfig.getWriteExecutorWaitStrategy(), preExecuteRunnable);
       case SIMPLE:
-        return new SimpleHoodieExecutor<>(inputItr, consumer, transformFunction, preExecuteRunnable);
+        return new SimpleHoodieExecutor<>(inputItr, consumer, transformFunction);
       default:
         throw new HoodieException("Unsupported Executor Type " + executorType);
     }

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/execution/TestSimpleExecutionInSpark.java

@@ -47,7 +47,6 @@
 import scala.Tuple2;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.mock;
@@ -99,14 +98,11 @@ public Integer finish() {
     SimpleHoodieExecutor<HoodieRecord, Tuple2<HoodieRecord, Option<IndexedRecord>>, Integer> exec = null;
 
     try {
-      exec = new SimpleHoodieExecutor(hoodieRecords.iterator(), consumer, getTransformFunction(HoodieTestDataGenerator.AVRO_SCHEMA), getPreExecuteRunnable());
-
+      exec = new SimpleHoodieExecutor(hoodieRecords.iterator(), consumer, getTransformFunction(HoodieTestDataGenerator.AVRO_SCHEMA));
 
       int result = exec.execute();
-      // It should buffer and write 100 records
+      // It should buffer and write 128 records
       assertEquals(128, result);
-      // There should be no remaining records in the buffer
-      assertFalse(exec.isRunning());
 
       // collect all records and assert that consumed records are identical to produced ones
       // assert there's no tampering, and that the ordering is preserved
@@ -175,12 +171,9 @@ public Integer finish() {
     SimpleHoodieExecutor<HoodieRecord, Tuple2<HoodieRecord, Option<IndexedRecord>>, Integer> exec = null;
 
     try {
-      exec = new SimpleHoodieExecutor(hoodieRecords.iterator(), consumer, getTransformFunction(HoodieTestDataGenerator.AVRO_SCHEMA), getPreExecuteRunnable());
+      exec = new SimpleHoodieExecutor(hoodieRecords.iterator(), consumer, getTransformFunction(HoodieTestDataGenerator.AVRO_SCHEMA));
       int result = exec.execute();
-      // It should buffer and write 100 records
       assertEquals(100, result);
-      // There should be no remaining records in the buffer
-      assertFalse(exec.isRunning());
 
       assertEquals(beforeRecord, afterRecord);
       assertEquals(beforeIndexedRecord, afterIndexedRecord);
@@ -223,7 +216,7 @@ public Integer finish() {
         };
 
     SimpleHoodieExecutor<HoodieRecord, Tuple2<HoodieRecord, Option<IndexedRecord>>, Integer> exec =
-        new SimpleHoodieExecutor(iterator, consumer, getTransformFunction(HoodieTestDataGenerator.AVRO_SCHEMA), getPreExecuteRunnable());
+        new SimpleHoodieExecutor(iterator, consumer, getTransformFunction(HoodieTestDataGenerator.AVRO_SCHEMA));
 
     final Throwable thrown = assertThrows(HoodieException.class, exec::execute,
         "exception is expected");

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

@@ -46,21 +46,17 @@ public class SimpleHoodieExecutor<I, O, E> implements HoodieExecutor<E> {
   // records iterator
   protected final Iterator<I> it;
   private final Function<I, O> transformFunction;
-  private final Runnable preExecuteRunnable;
-  private final AtomicBoolean isWriteDone = new AtomicBoolean(false);
-  private final AtomicBoolean isShutdown = new AtomicBoolean(false);
 
   public SimpleHoodieExecutor(final Iterator<I> inputItr, HoodieConsumer<O, E> consumer,
-                              Function<I, O> transformFunction, Runnable preExecuteRunnable) {
-    this(inputItr, Option.of(consumer), transformFunction, preExecuteRunnable);
+                              Function<I, O> transformFunction) {
+    this(inputItr, Option.of(consumer), transformFunction);
   }
 
   public SimpleHoodieExecutor(final Iterator<I> inputItr, Option<HoodieConsumer<O, E>> consumer,
-                              Function<I, O> transformFunction, Runnable preExecuteRunnable) {
+                              Function<I, O> transformFunction) {
     this.it = inputItr;
     this.consumer = consumer;
     this.transformFunction = transformFunction;
-    this.preExecuteRunnable = preExecuteRunnable;
   }
 
   /**
@@ -72,12 +68,7 @@ public E execute() {
 
     try {
       LOG.info("Starting consumer, consuming records from the records iterator directly");
-      preExecuteRunnable.run();
       while (it.hasNext()) {
-        if (isShutdown.get()) {
-          LOG.warn("Call shutdown while getting new entries, stop consuming.");
-          break;
-        }
         O payload = transformFunction.apply(it.next());
         consumer.get().consume(payload);
       }
@@ -86,22 +77,17 @@ public E execute() {
     } catch (Exception e) {
       LOG.error("Error consuming records in SimpleHoodieExecutor", e);
       throw new HoodieException(e);
-    } finally {
-      isWriteDone.set(true);
     }
   }
 
   @Override
   public void shutdownNow() {
-    isShutdown.set(true);
+    // no-op
   }
 
   @Override
   public boolean awaitTermination() {
-    return isWriteDone.get();
-  }
-
-  public boolean isRunning() {
-    return it.hasNext();
+    // no-op
+    return true;
   }
 }

hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/execution/benchmark/BoundInMemoryExecutorBenchmark.scala

@@ -66,6 +66,8 @@ object BoundInMemoryExecutorBenchmark extends HoodieBenchmarkBase {
   }
 
   /**
+   * This benchmark has been run w/ unconstrained parallelism which is beneficial to Disruptor more than it's for Simple
+   *
    * OpenJDK 64-Bit Server VM 1.8.0_342-b07 on Linux 5.10.62-55.141.amzn2.x86_64
    * Intel(R) Xeon(R) Platinum 8259CL CPU @ 2.50GHz
    * COW Ingestion:                            Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative