[MINOR] improve comaction operator shuffle

[xicm]

Change Logs

I found a skew in the flink compaction task.

Compaction task shuffle the compaction operation with keyby, A series of hash mods results in skew.

One file group on compaction operator, so we don't need keyby, just rebalance is OK.

This is the result of rebanlace.

Impact

none

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

none

Documentation Update

none

Contributor's checklist

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

[xicm]

One operator one file group, rebalance is ok.
This is what we do in HoodieFlinkCompactor

hudi/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java

Lines 289 to 303 in b7e11b1

	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	env.addSource(new CompactionPlanSourceFunction(compactionPlans, conf))
	.name("compaction_source")
	.uid("uid_compaction_source")
	.rebalance()
	.transform("compact_task",
	TypeInformation.of(CompactionCommitEvent.class),
	new CompactOperator(conf))
	.setParallelism(compactionParallelism)
	.addSink(new CompactionCommitSink(conf))
	.name("compaction_commit")
	.uid("uid_compaction_commit")
	.setParallelism(1)
	.getTransformation()
	.setMaxParallelism(1);

[danny0405]

Is rebanlance deterministic for multiple concurrent tasks, when there are task failover and retries, two task may has the risk of sharing a common compaction file group.

[danny0405]

Did you test it offline already?

[xicm]

I tested on my cluster. But I didn't test the failed and retry case.

[xicm]

Offline conpaction uses rebalance.

[hudi-bot]

CI report:

• 039fbc7 Azure: SUCCESS Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

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