[HUDI-4329] Add separate control for Flink compaction operation sync/async mode

[chenshzh]

Change Logs

Add separate control for Flink compaction operation sync/async mode.

Details in https://issues.apache.org/jira/projects/HUDI/issues/HUDI-4329

Problem Review

The compact operation sync/async in CompactionFunction is now controlled by FlinkOptions#COMPACTION_ASYNC_ENABLED

  public CompactFunction(Configuration conf) {
    this.conf = conf;
    this.asyncCompaction = StreamerUtil.needsAsyncCompaction(conf);
  }

While in fact it cannot be switched to sync mode because the pipeline defined by sync compaction will only include the clean but not compact operators.

  // compaction
  if (StreamerUtil.needsAsyncCompaction(conf)) {
    return Pipelines.compact(conf, pipeline);
  } else {
    return Pipelines.clean(conf, pipeline);
  }

Improvement

Add another separate control switch for compaction operation sync/async mode.

Impact

add sync compaction switch for MoR compaction.

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

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

Documentation Update

Add compaction.operation.async.enabled: used to turn on the synchronous compaction operation

The original compaction.async.enabled: used to turn on the compaction process.

Contributor's checklist

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

[chenshzh]

@danny0405 And if it's convenient, pls also check out this. It seems that it has been for some time. Thanks!

[danny0405]

Why switching to a new option key, what's the purpose here ?

[danny0405]

We already support sync compaction for bounded source, it should be fine for async compaction for streaming source ?

[chenshzh]

Normally we use compaction.async.enabled to turn on compaction. But we could not make it sync because it's already been true.

    if (asyncCompaction) {
      // executes the compaction task asynchronously to not block the checkpoint barrier propagate.
      executor.execute(
          () -> doCompaction(instantTime, compactionOperation, collector, reloadWriteConfig()),
          (errMsg, t) -> collector.collect(new CompactionCommitEvent(instantTime, compactionOperation.getFileId(), taskID)),
          "Execute compaction for instant %s from task %d", instantTime, taskID);
    } else {
      // executes the compaction task synchronously for batch mode.
      LOG.info("Execute compaction for instant {} from task {}", instantTime, taskID);
      doCompaction(instantTime, compactionOperation, collector, writeClient.getConfig());
    }

support sync compaction for bounded source

We will use sync compaction mode for unbounded source in some scenarios. And actually the bounded source sync compaction seems weird. It use compaction.async.enabled true to turn on compaction, and then switch it to fasle for sync mode.

     // compaction
      if (OptionsResolver.needsAsyncCompaction(conf)) {   // here FlinkOptions.COMPACTION_ASYNC_ENABLED decides that we need compaction
        // use synchronous compaction for bounded source.
        if (context.isBounded()) {
          conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false); // we come here because it is true, and it's so weird to turn  it false, actually we just want the operation to be executed sync.
        }
        return Pipelines.compact(conf, pipeline);
      } else {
        return Pipelines.clean(conf, pipeline);
      }

[chenshzh]

@danny0405 pls help see whether there are any more problems here?

[danny0405]

And actually the bounded source sync compaction seems weird. It use compaction.async.enabled true to turn on compaction, and then switch it to fasle for sync mode.

I agree the logic here is a little weird, we can refactor it though,
the bounded source sync compaction is meaningful, especially when people do batch ingestion for mor table, they do not what a separate compaction job running again.

[chenshzh]

keep sync compaction as an option for unbounded source.

And in a whole for hudi feature, you might also agree that users should be provided sync compaction option for unbounded source, no matter the above mentioned scenarios?

[chenshzh]

@danny0405 do we have any other questions here?

[danny0405]

and at the same time we use async thread to execute compaction async and collect the result compaction msgs, the output.collector will become thread unsafe

The output collector is thread unsafe here, let's fix it to be thread safe then ~

[chenshzh]

OK, but that will be another issue, i think.
What about this pr ? We have discussed much about its necessity to keep sync compaction option for unbounded source.

[danny0405]

The watermark conflicts issue has been addressed in #8379, while I still deem this PR as valid, the suggestions:

1. rename compaction.async.enabled to compaction.enabled
2. store the isBounded option in the conf to switch to sync compaction for bounded source.

[hudi-bot]

CI report:

• 0ca6be8 UNKNOWN
• d520045 Azure: CANCELED

Bot commands

@hudi-bot supports the following commands:

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

[yihua]

@danny0405 is this still needed for Hudi Flink?

[danny0405]

Yeah, let's move it to 1.0 release.