[HUDI-5023] Cleaning up `QueueBasedExecutor` impls #7238

alexeykudinkin · 2022-11-18T04:45:52Z

Change Logs

This is a follow-up PR for #5416 that is

Further cleaning up some of the historically inherited artifacts
Replacing remaining usages of the BIMQ, w/ QueueBasedExecutorFactory
Adding missing deps to bundles

This change is a precursor for both #7245 and #7174

Impact

No impact

Risk level (write none, low medium or high below)

Low

Documentation Update

N/A

Contributor's checklist

Read through contributor's guide
Change Logs and Impact were stated clearly
Adequate tests were added if applicable
CI passed

zhangyue19921010 · 2022-11-18T06:42:14Z

Ack Will finish my review this week.

alexeykudinkin · 2022-11-18T23:37:00Z

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

-   * Start all producers at once.
-   */
-  @Override
-  public CompletableFuture<Void> startProducers() {


This method (w/ modifications) have been moved into base class to be shared across impls

alexeykudinkin · 2022-11-18T23:38:49Z

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

+  // Consumer
+  protected final Option<HoodieConsumer<O, E>> consumer;
+
+  public BaseHoodieQueueBasedExecutor(List<HoodieProducer<I>> producers,


Ideas underpinning this class were

Abstract common logic across impls

Make sure all concurrency is handled exclusively in the base class (impl should just write sync code of how to produce/consume to/from the queue)

Make sure all resource lifecycle handling happens in the base class

Nice work here!

alexeykudinkin · 2022-11-18T23:39:46Z

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

- * HoodieExecutorBase holds common elements producerExecutorService, consumerExecutorService, producers and a single consumer.
- * Also HoodieExecutorBase control the lifecycle of producerExecutorService and consumerExecutorService.
- */
-public abstract class HoodieExecutorBase<I, O, E> implements HoodieExecutor<I, O, E> {


This class has been replaced by BaseHoodieQueueBasedExecutor

alexeykudinkin · 2022-11-18T23:39:58Z

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

-/**
- * IteratorBasedHoodieMessageQueue implements HoodieMessageQueue with Iterable
- */
-public abstract class HoodieIterableMessageQueue<I, O> implements HoodieMessageQueue<I, O>, Iterable<O> {


This is not needed anymore

alexeykudinkin · 2022-11-18T23:40:20Z

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

-/**
- * Consume entries from queue and execute callback function.
- */
-public abstract class IteratorBasedQueueConsumer<I, O> implements HoodieConsumer<I, O> {


This is not needed anymore

alexeykudinkin · 2022-11-18T23:41:23Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java

@@ -208,19 +205,33 @@ public BoundedMemoryRecords(
        Schema logSchema,
        Configuration hadoopConf,
        org.apache.flink.configuration.Configuration flinkConf) {
+      HoodieUnMergedLogRecordScanner.Builder scannerBuilder = HoodieUnMergedLogRecordScanner.newBuilder()


This refactoring is necessary to avoid exposing internals of the executor (queue), instead relying on the provided interfaces to configure it

zhangyue19921010 · 2022-11-21T05:43:09Z

@hudi-bot run azure

zhangyue19921010

Nice Work!
Just left a few minor comments

zhangyue19921010 · 2022-11-23T05:49:48Z

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

+/**
+ * HoodieExecutorBase holds common elements producerExecutorService, consumerExecutorService, producers and a single consumer.
+ * Also HoodieExecutorBase control the lifecycle of producerExecutorService and consumerExecutorService.
+ */
+public abstract class BaseHoodieQueueBasedExecutor<I, O, E> implements HoodieExecutor<I, O, E> {
+


nit: HoodieExecutorBase => BaseHoodieQueueBasedExecutor

zhangyue19921010 · 2022-11-23T06:37:34Z

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

+  public final CompletableFuture<Void> startProducingAsync() {
+    return allOf(producers.stream()
+        .map(producer -> CompletableFuture.supplyAsync(() -> {
+          doProduce(queue, producer);
+          return (Void) null;
+        }, producerExecutorService))
+        .collect(Collectors.toList())
+    )
+        .thenApply(ignored -> (Void) null)
+        .whenComplete((result, throwable) -> {
+          // Regardless of how producing has completed, we have to close producers
+          // to make sure resources are properly cleaned up
+          producers.forEach(HoodieProducer::close);
+          // Mark production as done so that consumer will be able to exit
+          queue.seal();
+        });
+  }


How about using

public final CompletableFuture<Void> startProducingAsync() { return CompletableFuture.allOf(producers.stream().map(producer -> { return CompletableFuture.supplyAsync(() -> { doProduce(queue, producer); return (Void) null; }, producerExecutorService); }).toArray(CompletableFuture[]::new)).whenComplete(new BiConsumer<Void, Throwable>() { @Override public void accept(Void aVoid, Throwable throwable) { // Regardless of how producing has completed, we have to close producers // to make sure resources are properly cleaned up producers.forEach(HoodieProducer::close); // Mark production as done so that consumer will be able to exit queue.seal(); } }); }

may be no need to do .thenApply(ignored -> (Void) null)

In #7245, i'm actually going to replace this w/ an allOf instance having slightly different semantic (allowing to cascade cancellations upon first encountered failure)

zhangyue19921010 · 2022-11-23T07:52:43Z

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

+      return producing.thenCombine(consuming, (aVoid, anotherVoid) -> null)
+          .whenComplete((ignored, throwable) -> {
+            // Close the queue to release the resources
+            queue.close();
+          })
+          .thenApply(ignored -> consumer.get().finish())
+          // Block until producing and consuming both finish
+          .join();
+    } catch (Exception e) {
+      throw new HoodieException(e);
+    }


It seems that TestBoundedInMemoryExecutorInSpark#testInterruptExecutor is blocked here which case ut timeout.
Maybe we can revisit the logic here about how to handle interrupt exception.

Yeah, join actually doesn't throw InterruptedException. Addressed

zhangyue19921010 · 2022-11-23T08:06:36Z

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

+  // Consumer
+  protected final Option<HoodieConsumer<O, E>> consumer;
+
+  public BaseHoodieQueueBasedExecutor(List<HoodieProducer<I>> producers,


Nice work here!

zhangyue19921010 · 2022-11-23T08:14:17Z

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

+  @Override
+  public void shutdownNow() {
+    producerExecutorService.shutdownNow();
+    consumerExecutorService.shutdownNow();


Maybe we need to close current queue into this shutdownNow func.

We actually close the queue w/in the execute method even in the presence of exceptions (which will cascade from shutdownNow if invoked). Idea here is that shutdownNow should only be used as last-ditch effort of (quickly!) shutting down spun up threads only

That make sense !

zhangyue19921010 · 2022-11-23T08:44:48Z

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

@@ -31,11 +32,18 @@

 public class QueueBasedExecutorFactory {


Maybe we can also rename QueueBasedExecutorFactory to ExecutorFactory because we will support simpleExecutor here which has no inner Queue

Agreed. Let's take this up in the PR introducing SimpleExecutor

zhangyue19921010 · 2022-12-02T03:43:18Z

@hudi-bot run azure

codope

@alexeykudinkin @zhangyue19921010 Do you plan to re-run the same benchmark as in #5416 for this PR as well?

packaging/hudi-flink-bundle/pom.xml

codope · 2022-12-02T06:58:15Z

hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeUnmergedRecordReader.java

@@ -149,6 +153,7 @@ public void close() throws IOException {

  @Override
  public float getProgress() throws IOException {
-    return Math.min(parquetReader.getProgress(), logRecordScanner.getProgress());
+    // TODO fix to reflect scanner progress


Why would we need that for "unmerged" record reader?

I think this is just tracking progress in terms of how many records were processed from the base-file vs logs

Synced up offline. This will be addressed in followup PR.

codope · 2022-12-02T07:07:27Z

hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeUnmergedRecordReader.java

+                  })
+                  .build();
+          // Scan all the delta-log files, filling in the queue
+          scanner.scan();


maybe assign scanner progress here, which can be used in L157.

Will take this up in #7245

zhangyue19921010 · 2022-12-02T13:46:14Z

@alexeykudinkin @zhangyue19921010 Do you plan to re-run the same benchmark as in #5416 for this PR as well?

Yeap, I run this benchmark several times based on this PR and the result are the same as before. Thanks for reminding!

zhangyue19921010

LTGM! Thanks for this GREAT work!

After UTs green maybe we can land it. Maybe we can rebase master #7346 and have another try.

…y` instead

Cleaned up `HoodieIterableMessageQueue`

…tind down the queue prematurely)

Fixed handling of the case when there's no consumer provide closing the queue prematurely

…e of DQ returning prematurely

hudi-bot · 2022-12-03T12:19:55Z

CI report:

ba03091 UNKNOWN
e4652c0 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

@hudi-bot run azure re-run the last Azure build

codope · 2022-12-07T04:51:56Z

hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeUnmergedRecordReader.java

@@ -149,6 +153,7 @@ public void close() throws IOException {

  @Override
  public float getProgress() throws IOException {
-    return Math.min(parquetReader.getProgress(), logRecordScanner.getProgress());
+    // TODO fix to reflect scanner progress


Synced up offline. This will be addressed in followup PR.

- Further cleaning up some of the historically inherited artifacts - Replacing remaining usages of the BIMQ, w/ QueueBasedExecutorFactory - Adding missing deps to bundles - Lifted up all concurrency management to `BaseHoodieQueueBasedExuecutor` - Fixed handling of the case when there's no consumer provide closing the queue prematurely

alexeykudinkin mentioned this pull request Nov 18, 2022

[HUDI-5023] Consuming records from Iterator directly instead of using inner message queue #7174

Merged

4 tasks

alexeykudinkin force-pushed the ak/exeq-fix branch from 6241a82 to eb19e94 Compare November 18, 2022 23:35

alexeykudinkin commented Nov 18, 2022

View reviewed changes

alexeykudinkin force-pushed the ak/exeq-fix branch from eb19e94 to e17b127 Compare November 22, 2022 06:17

zhangyue19921010 reviewed Nov 23, 2022

View reviewed changes

alexeykudinkin force-pushed the ak/exeq-fix branch 2 times, most recently from 5e9080b to 3532ca4 Compare November 30, 2022 17:37

alexeykudinkin changed the title ~~[HUDI-3963] Cleaning up QueueBasedExecutor impls~~ [HUDI-5023] Cleaning up QueueBasedExecutor impls Nov 30, 2022

alexeykudinkin force-pushed the ak/exeq-fix branch 2 times, most recently from 0c62553 to a76c39f Compare December 1, 2022 16:25

codope added priority:critical production down; pipelines stalled; Need help asap. writer-core Issues relating to core transactions/write actions labels Dec 2, 2022

codope assigned codope and zhangyue19921010 Dec 2, 2022

codope reviewed Dec 2, 2022

View reviewed changes

zhangyue19921010 approved these changes Dec 2, 2022

View reviewed changes

Alexey Kudinkin added 10 commits December 2, 2022 19:17

Fixed incorrect use of CountDownLatch

767ddd8

Removed unnecessary casting

4ebf875

Adding logs

77d7e9d

Unified getResult and finish in one API

a4672e0

Simplifying DisruptorExecutor

d81602a

BoundedInMemoryQueueIterable > BoundedInMemoryQueue

14f2956

Lifted all queue lifecycle management to HoodieExecutorBase

326ed2c

Tidying up

a267452

Rebased direct users of the BIMQ to rely on `QueueBasedExecutorFactor…

aee1b84

…y` instead

Tidying up

280ff3a

Alexey Kudinkin added 19 commits December 2, 2022 19:17

Close the queue after processing is done

034c123

Cleaning up superfluous APIs

ef42ebf

Adding missing deps to bundles

b286f0f

Tidying up

09ca087

Moved actual queue iteration into BoundedInMemoryQueueExecutor

d3726e4

Cleaned up IteratorBasedQueueConsumer

9f9aad6

Revisited BoundedInMemoryExecutor bootstrapping w/o consumer;

25218aa

Cleaned up `HoodieIterableMessageQueue`

Streamline and unify producing handling w/in Queue-based executors

ec5b19c

Abstracted common producing seq

9028449

Delineated sealing/closing of the queue (addressing the issue of shut…

30c4d85

…tind down the queue prematurely)

Tidying up

4214901

Lifted up all concurrency management to BaseHoodieQueueBasedExuecutor;

7779c5f

Fixed handling of the case when there's no consumer provide closing the queue prematurely

Decouple consuming seq from returning the result, to address the issu…

c2b9d85

…e of DQ returning prematurely

Relocated Disruptor queue setup under doConsume

4310334

Tidying up

a3c2c11

Replace join w/ get to be able to handle InterruptedException

8c2a34a

Simplify the test

ccf65c1

Tidying up

d05236b

Fixed tests

e4652c0

alexeykudinkin force-pushed the ak/exeq-fix branch from a76c39f to e4652c0 Compare December 3, 2022 03:18

codope approved these changes Dec 7, 2022

View reviewed changes

codope merged commit 20eea46 into apache:master Dec 7, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[HUDI-5023] Cleaning up `QueueBasedExecutor` impls #7238

[HUDI-5023] Cleaning up `QueueBasedExecutor` impls #7238

alexeykudinkin commented Nov 18, 2022 •

edited

zhangyue19921010 commented Nov 18, 2022

alexeykudinkin Nov 18, 2022

alexeykudinkin Nov 18, 2022

zhangyue19921010 Nov 23, 2022

alexeykudinkin Nov 18, 2022

alexeykudinkin Nov 18, 2022

alexeykudinkin Nov 18, 2022

alexeykudinkin Nov 18, 2022

zhangyue19921010 commented Nov 21, 2022

zhangyue19921010 left a comment

zhangyue19921010 Nov 23, 2022

zhangyue19921010 Nov 23, 2022

alexeykudinkin Nov 30, 2022 •

edited

zhangyue19921010 Nov 23, 2022

alexeykudinkin Nov 30, 2022

zhangyue19921010 Nov 23, 2022

zhangyue19921010 Nov 23, 2022

alexeykudinkin Nov 30, 2022

zhangyue19921010 Dec 2, 2022

zhangyue19921010 Nov 23, 2022

alexeykudinkin Nov 30, 2022

zhangyue19921010 commented Dec 2, 2022

codope left a comment

codope Dec 2, 2022

alexeykudinkin Dec 3, 2022

codope Dec 7, 2022

codope Dec 2, 2022

alexeykudinkin Dec 3, 2022

zhangyue19921010 commented Dec 2, 2022 •

edited

zhangyue19921010 left a comment •

edited

hudi-bot commented Dec 3, 2022

codope Dec 7, 2022

[HUDI-5023] Cleaning up QueueBasedExecutor impls #7238

[HUDI-5023] Cleaning up QueueBasedExecutor impls #7238

Conversation

alexeykudinkin commented Nov 18, 2022 • edited

Change Logs

Impact

Risk level (write none, low medium or high below)

Documentation Update

Contributor's checklist

zhangyue19921010 commented Nov 18, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhangyue19921010 commented Nov 21, 2022

zhangyue19921010 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alexeykudinkin Nov 30, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhangyue19921010 commented Dec 2, 2022

codope left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhangyue19921010 commented Dec 2, 2022 • edited

zhangyue19921010 left a comment • edited

Choose a reason for hiding this comment

hudi-bot commented Dec 3, 2022

CI report:

Choose a reason for hiding this comment

[HUDI-5023] Cleaning up `QueueBasedExecutor` impls #7238

[HUDI-5023] Cleaning up `QueueBasedExecutor` impls #7238

alexeykudinkin commented Nov 18, 2022 •

edited

alexeykudinkin Nov 30, 2022 •

edited

zhangyue19921010 commented Dec 2, 2022 •

edited

zhangyue19921010 left a comment •

edited