[flink] respect flink io tmp multi dirs to fix disk exhaustion#3617
Open
zuston wants to merge 4 commits into
Open
[flink] respect flink io tmp multi dirs to fix disk exhaustion#3617zuston wants to merge 4 commits into
zuston wants to merge 4 commits into
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
This PR addresses #3616 by improving how Fluss derives client.scanner.io.tmpdir from Flink’s io.tmp.dirs when running on YARN, including support for multiple tmp directories to avoid concentrating all temporary files on a single disk.
Changes:
- Add task-index aware tmp-dir selection and split handling for multiple Flink
TMP_DIRSentries. - Update Flink source readers to pass subtask index so different subtasks can use different tmp dirs.
- Extend unit tests to cover comma-separated and path-separator-separated tmp dir lists.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 3 comments.
| File | Description |
|---|---|
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtils.java | Adds multi-dir parsing and task-index based tmp dir selection logic. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java | Passes subtask index into tmp dir selection. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java | Passes subtask index into tmp dir selection for tiering reader. |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtilTest.java | Adds coverage for multiple tmp dir formats and updates expectations. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
233
to
237
| private static String[] splitPaths(@Nonnull String separatedPaths) { | ||
| return separatedPaths.length() > 0 | ||
| ? separatedPaths.split(",|" + File.pathSeparator) | ||
| : new String[0]; | ||
| } |
Comment on lines
107
to
+108
| .isEqualTo("/flink_tmp_dir/fluss"); | ||
| assertThat(FlinkConnectorOptionsUtils.getClientScannerIoTmpDir(flussConfig, flinkConfig, 1)) |
beryllw
reviewed
Jul 8, 2026
wuchong
reviewed
Jul 8, 2026
beryllw
reviewed
Jul 8, 2026
| @@ -290,7 +290,8 @@ public SourceReader<OUT, SourceSplitBase> createReader(SourceReaderContext conte | |||
|
|
|||
| flussConf.set( | |||
Contributor
There was a problem hiding this comment.
Just a personal question, might be missing something here — since flussConf is a member field of the source, calling set seems to mutate it in place, and this config eventually gets passed down to the Connection and shared by the downstream scanners. So I'm wondering whether multiple readers reusing the same instance could end up sharing the same config. Would it be safer to copy it first before setting？
Curious how you see it @luoyuxia ?
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Member
Author
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
Linked issue: close #3616
Brief change log
respect flink io tmp multi dirs to fix disk exhaustion
Tests
API and Format
Documentation