[HUDI-5488] Make sure Disrupt queue start first, then insert records

[boneanxs]

Change Logs

Describe context and summary for this change. Highlight if any code was copied.
We must to make sure to set up Disruptor's queue first, then producer can insert records to the queue. But currently we have no idea which thread start first, so this pr tries to fix it.

CompletableFuture<Void> consuming = startConsumingAsync();
CompletableFuture<Void> producing = startProducingAsync();

Also, I think the test TestDisruptorExecutionInSpark#testExecutor and TestDisruptorMessageQueue#testRecordReading failures relate to this bug.

Impact

Describe any public API or user-facing feature change or any performance impact.
none

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

If medium or high, explain what verification was done to mitigate the risks.

none

Documentation Update

Describe any necessary documentation update if there is any new feature, config, or user-facing change

• The config description must be updated if new configs are added or the default value of the configs are changed
• Any new feature or user-facing change requires updating the Hudi website. Please create a Jira ticket, attach the
  ticket number here and follow the instruction to make
  changes to the website.

Contributor's checklist

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

[boneanxs]

@alexeykudinkin @zhangyue19921010 Could you please help to take a look?

[zhangyue19921010]

Nice catch @boneanxs !
We do have to make sure disruptor is set up before we are consuming records.

But do we need to have a synchronized check for each record consuming(performance concern)? Or we do a setup() step in BaseHoodieQueueBasedExecutor#execute before startConsumingAsync and startProducingAsync

Hi @alexeykudinkin what do u think :)

[zhangyue19921010]

For example something like this

public class DisruptorExecutor<I, O, E> extends BaseHoodieQueueBasedExecutor<I, O, E> {

  public DisruptorExecutor(final Option<Integer> bufferSize, final Iterator<I> inputItr,
                           HoodieConsumer<O, E> consumer, Function<I, O> transformFunction, Option<String> waitStrategy, Runnable preExecuteRunnable) {
    this(bufferSize, Collections.singletonList(new IteratorBasedQueueProducer<>(inputItr)), Option.of(consumer),
        transformFunction, waitStrategy, preExecuteRunnable);
  }

  public DisruptorExecutor(final Option<Integer> bufferSize, List<HoodieProducer<I>> producers,
                           Option<HoodieConsumer<O, E>> consumer, final Function<I, O> transformFunction,
                           final Option<String> waitStrategy, Runnable preExecuteRunnable) {
    super(producers, consumer, new DisruptorMessageQueue<>(bufferSize, transformFunction, waitStrategy, producers.size(), preExecuteRunnable), preExecuteRunnable);
  }

  @Override
  protected void doConsume(HoodieMessageQueue<I, O> queue, HoodieConsumer<O, E> consumer) {
    // no-op
    // will do consuming actions in disruptor
  }

  @Override
  public void setup() {
    DisruptorMessageQueue<I, O> disruptorQueue = (DisruptorMessageQueue<I, O>) queue;
    // Before we start producing, we need to set up Disruptor's queue
    disruptorQueue.setHandlers(consumer.get());
    disruptorQueue.start();
  }
}

and

  public void setup(){}

  /**
   * Main API to run both production and consumption.
   */
  @Override
  public E execute() {
    try {
      checkState(this.consumer.isPresent());
      setup();
      // Start consuming/producing asynchronously
      CompletableFuture<Void> consuming = startConsumingAsync();
      CompletableFuture<Void> producing = startProducingAsync();

      // NOTE: To properly support mode when there's no consumer, we have to fall back
      //       to producing future as the trigger for us to shut down the queue
      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
          .get();
    } catch (Exception e) {
      if (e instanceof InterruptedException) {
        // In case {@code InterruptedException} was thrown, resetting the interrupted flag
        // of the thread, we reset it (to true) again to permit subsequent handlers
        // to be interrupted as well
        Thread.currentThread().interrupt();
      }

      throw new HoodieException(e);
    }
  }

[boneanxs]

But do we need to have a synchronized check for each record consuming(performance concern)?

This fix won't do the synchronize check for each record, it will only block the consuming when the queue is not start yet.

But I think adding a new setup() method is also an appropriate way. Pending this for a while if @alexeykudinkin has other suggestion. Or I'll add the setup() to fix it.

[biaoma-ty]

typo in title

[zhangyue19921010]

This PR may relate to https://issues.apache.org/jira/browse/HUDI-5369

[boneanxs]

Gentle ping @alexeykudinkin

[alexeykudinkin]

@boneanxs @zhangyue19921010

We should do what @zhangyue19921010 suggested in his review

1. Introduce empty setup method in the base class
2. Move current contents of the doConsume for Disruptor into setup
3. Keep doConsume for Disruptor empty

[alexeykudinkin]

I don't think we need this, it would throw anyway if the queue has not been started yet

[boneanxs]

We assert in insertRecord that the queue is started

Here just assert the queue is started, in case others use this queue without calling the start

[alexeykudinkin]

@boneanxs we need to keep in mind that this lies on the hot-path of being executed for every record, so we should keep the bar really high in terms of what we allow to be executed for every record.

That being said, i also see your point of adding it to make sure it would fail definitively unless queue has been started

[alexeykudinkin]

Let's keep it, please rename it to isStarted

[boneanxs]

@hudi-bot run azure

[zhangyue19921010]

@hudi-bot run azure

[boneanxs]

The test failure should not relate to this, I can run it successfully in local side:

[boneanxs]

@hudi-bot run azure

[boneanxs]

@hudi-bot run azure

[alexeykudinkin]

@boneanxs we need to keep in mind that this lies on the hot-path of being executed for every record, so we should keep the bar really high in terms of what we allow to be executed for every record.

That being said, i also see your point of adding it to make sure it would fail definitively unless queue has been started

[alexeykudinkin]

Let's keep it, please rename it to isStarted

[zhangyue19921010]

LGTM. We can land it after this nit(Alex mentioned) changed. Thanks for your contribution!

[hudi-bot]

CI report:

• a94ec9c Azure: FAILURE Azure: SUCCESS
• 80fedf4 Azure: PENDING

Bot commands

@hudi-bot supports the following commands:

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

[alexeykudinkin]

Landing this since there's no material changes since last revision, and CI passed previously