[HUDI-6134] Prevent two clean run concurrently in flink #8568
Conversation
|
Can you elaborate the details a little, the |
If clean called in |
If the operator is not opened, how could it execute the |
In |
| this.isCleaning = true; | ||
| try { | ||
| this.writeClient.clean(instantTime); | ||
| } catch (Throwable throwable) { |
There was a problem hiding this comment.
We need to execute the cleaning no matter whether async cleaning is true or false.
And this is a NonThrownExecutor, there is no need to catch the exception.
There was a problem hiding this comment.
Right. Removed the catch clause and ASYNC_ENABLED condition, and add !isCleaning condition before clean function calls in subclass ClusteringCommitSink and CompactionCommitSink.
Pls take another look, thanks
hbgstc123
force-pushed
the
HUDI-6134_prevent_clean_run_concurrently_in_flink
branch
from
6926fef
to
708a1e0
Compare
| @@ -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"); | |||
There was a problem hiding this comment.
What's the matter if there are multiple cleaning tasks runing here?
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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
There was a problem hiding this comment.
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?
There was a problem hiding this comment.
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?
danny0405
changed the title
danny0405
changed the title
Co-authored-by: hbg <bingeng.huang@shopee.com>
Co-authored-by: hbg <bingeng.huang@shopee.com>
Change Logs
prevent two clean run concurrently in flink.
Impact
no
Risk level (write none, low medium or high below)
none
Documentation Update
none
Contributor's checklist