Skip to content

[HUDI-6134] Prevent two clean run concurrently in flink #8568

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
danny0405 merged 1 commit into from
May 16, 2023
Merged

[HUDI-6134] Prevent two clean run concurrently in flink #8568

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/CleanFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ public class CleanFunction<T> extends AbstractRichFunction

private NonThrownExecutor executor;

private volatile boolean isCleaning;
protected volatile boolean isCleaning;

public CleanFunction(Configuration conf) {
this.conf = conf;
Expand All @@ -64,7 +64,14 @@ public void open(Configuration parameters) throws Exception {
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for cleaning finish");
executor.execute(() -> {
this.isCleaning = true;
try {
this.writeClient.clean(instantTime);
} finally {
this.isCleaning = false;
}
}, "wait for cleaning finish");
}

@Override
Expand Down
2 changes: 1 addition & 1 deletion ...source/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringCommitSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,7 +179,7 @@ private void doCommit(String instant, HoodieClusteringPlan clusteringPlan, List<
TableServiceType.CLUSTER, writeMetadata.getCommitMetadata().get(), table, instant);

// whether to clean up the input base parquet files used for clustering
if (!conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
if (!conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && !isCleaning) {
LOG.info("Running inline clean");
Copy link
Contributor

@danny0405 danny0405 Apr 26, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What's the matter if there are multiple cleaning tasks runing here?

Copy link
Contributor Author

@hbgstc123 hbgstc123 Apr 26, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Multiple cleaning tasks maybe running the same clean instant, it's unnecessary.
And now multi clean can leave .hoodie/20230425173418352.clean.tmp this tmp file in timeline because the slower clean task will fail to rename this tmp file to the final 20230425173418352.clean because it will be there created by an earlier clean task.
We can try fix the rename issue in HoodieActiveTimeline.transitionState too, delete the tmp file when rename is unsuccessful.

Copy link
Contributor

@danny0405 danny0405 Apr 26, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Multiple cleaning tasks maybe running the same clean instant, it's unnecessary.

That's not possible, cleaning service may have clean plan, one file can only belongs to one cleaning plan.

Copy link
Contributor Author

@hbgstc123 hbgstc123 May 16, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

First I see the clean.tmp file in a flink write hudi job with online compaction,
recently I see the clean tmp file in timeline again in a flink job with offline compaction.
I write an issue about it:
#8726

Copy link
Contributor

@danny0405 danny0405 May 16, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Multiple cleaning tasks maybe running the same clean instant

Can you explain why the same cleaning instant is executed by multiple tasks? You mean the cleaning service would try to execute the existing pending cleaning instant? Can you show us the code there?

Copy link
Contributor

@danny0405 danny0405 May 16, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

After some analysis I find that there is an option hoodie.clean.allow.multiple for multiple writer cleaning, and by default it is true.

I also find that the clean action executor would try to clean all the pending cleaning instants on the timeline if hoodie.clean.allow.multiple is enabled.

But in flink streaming, the cleaning is scheduled and executed in a single worker thread pool, that means at most 1 cleaning task is running for streaming pipeline.

But there is possibility for batch ingestion job, the #open method and #doCommit method can trigger the cleaning in separeate threads. Is that your case here?

this.writeClient.clean();
}
Expand Down
2 changes: 1 addition & 1 deletion ...atasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -175,7 +175,7 @@ private void doCommit(String instant, Collection<CompactionCommitEvent> events)
this.writeClient.commitCompaction(instant, metadata, Option.empty());

// Whether to clean up the old log file when compaction
if (!conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
if (!conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && !isCleaning) {
this.writeClient.clean();
}
}
Expand Down