You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "LadyForest (via GitHub)" <gi...@apache.org> on 2024/02/27 07:50:15 UTC
[PR] [FLINK-29114][table-planner][filesystem][test] Fix unstable TableSourceITCase#testTableHintWithLogicalTableScanReuse [flink]
LadyForest opened a new pull request, #24390:
URL: https://github.com/apache/flink/pull/24390
## What is the purpose of the change
This pull request aims to fix the unstable test case `TableSourceITCase#testTableHintWithLogicalTableScanReuse`.
## Brief change log
Add a random UUID to `FileSystemTableSink#toStagingPath` to avoid path conflicts.
## Verifying this change
You can revert this change and run `TableSourceITCase#testTableHintWithLogicalTableScanReuse` in the IDE; it should fail after several times. Then, apply this change and run again. The failure should not appear anymore.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1505976015
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
Review Comment:
Thanks for clarifying things. I tried to come up with a proposal for a test. But to be fair, even that one is not perfect (you want to avoid weakening the accessibility of production code for tests). It hard to achieve in other ways because of the way the classes are structured (as you already pointed out). Doing it properly would require a larger refactoring.
Have a look at LadyForest/flink#1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1512103738
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/utils/PathUtils.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.file.table.utils;
+
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.UUID;
+
+/** Path utils for file system. */
+public class PathUtils {
+
+ public static Path getStagingPath(Path path) {
Review Comment:
Hi @XComp, I'm afraid that this way also violates the arch unit test.
Here's what I tested in the local.
```java
public class PathUtils {
public static Path getStagingPath(Path path) {
// Add a random UUID to prevent multiple sinks from sharing the same staging dir.
// Please see FLINK-29114 for more details
return getStagingPath(
path,
() ->
String.format(
".staging_%d_%s", System.currentTimeMillis(), UUID.randomUUID()));
}
@VisibleForTesting
static Path getStagingPath(Path path, Supplier<String> suffixSupplier) {
Path stagingDir = path.suffix(suffixSupplier.get());
try {
FileSystem fs = stagingDir.getFileSystem();
Preconditions.checkState(
!fs.exists(stagingDir), "Staging dir %s already exists", stagingDir);
fs.mkdirs(stagingDir);
return stagingDir;
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
```
Run `ArchitectureTest` to get the following stacktrace. (The first rule has been violated before, so let's leave that.)
```
java.lang.AssertionError: Architecture Violation [Priority: MEDIUM] - Rule 'Connector production code must depend only on public API when outside of connector packages' was violated (2 times):
Method <org.apache.flink.connector.file.table.utils.PathUtils.getStagingPath(org.apache.flink.core.fs.Path, java.util.function.Supplier)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.String, [Ljava.lang.Object;)> in (PathUtils.java:48)
Method <org.apache.flink.connector.file.table.utils.PathUtils.getStagingPath(org.apache.flink.core.fs.Path, java.util.function.Supplier)> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (PathUtils.java:0)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1505822872
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
Review Comment:
The reason I added tests in `FileSystemOutputFormatTest` is that the issue is triggered when the `FileSystemOutputFormat#finalizeGlobal` method is called (triggered by the final block).
If we add tests only for `FileSystemTableSink`, we won't be able to reproduce the issue and verify the fix. Do you think it would be a good idea if I made `FileSystemTableSink#toStagingPath` a static method, added the `VisibleForTesting` annotation, and called it in `FileSystemOutputFormatTest`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on PR #24390:
URL: https://github.com/apache/flink/pull/24390#issuecomment-1970712539
Actually, after looking at the code for `HiveTableSink` https://github.com/apache/flink/blob/9e81177e44d63501b360b1a246a16ea3faecb548/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSink.java#L791, I am even more convinced that we should move the logic for generating the staging directory to the internal of the `OutputFormat`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1521195360
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -37,24 +39,44 @@
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
+import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
-import java.util.TreeMap;
-import java.util.concurrent.atomic.AtomicReference;
+import java.util.UUID;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.entry;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link FileSystemOutputFormat}. */
class FileSystemOutputFormatTest {
- @TempDir private java.nio.file.Path tmpPath;
@TempDir private java.nio.file.Path outputPath;
+ @TempDir private java.nio.file.Path stagingBaseDir;
+
+ private java.nio.file.Path stagingPath;
Review Comment:
The reason to introduce both`stagingBaseDir` and `stagingPath` is that
1. We cannot directly use `@TempDir` to annotate `stagingPath`. Junit5 will automatically create a temp directory, and thus the test will fail at `Preconditions.checkState`.
2. The reason to introduce `@TempDir stagingBaseDir`(rather than reuse `outputPath`) is to decouple the test logic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1507195791
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -104,15 +113,101 @@ void testNonPartition() throws Exception {
.containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testStagingDirBehavior(boolean shareStagingDir) throws Exception {
Review Comment:
> Generally speaking, unit tests should be as small as possible to focus on the specific aspect of the class contract (in this case `FileSystemTableSink#toStagingPath` not returning the same folder twice. This test adds quite a bit of extra context here which might make it harder to grasp the intention for code readers. What's your opinion on that?
+1 with your perspective on unit testing. As I previously mentioned, in this case, constructing a UT is quite challenging, which subtly suggests to me that there may be some design issues with the current code. Take the staging dir, for example, if it needs to be passed from the `FileSystemTableSink` to the `OutputFormat` at compile-time, then a proper validation should be done here. Conversely, perhaps the `OutputFormat` could decide on the stagingDir itself, which would make the testing more targeted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1505790506
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
Review Comment:
yes, confirm, it seems the current tests do not invoke `FileSystemTableSink#toStagingPath` in either way
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1505879003
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
Review Comment:
How can we effectively test and reproduce a sporadic runtime issue caused by improperly configured parameters at the compile phase in the unit test case? Do you have any good suggestions?🤔
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on PR #24390:
URL: https://github.com/apache/flink/pull/24390#issuecomment-1991461563
Hi [XComp](https://github.com/XComp), thanks so much for the time and effort you've put into the review and discussion :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1505770714
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
Review Comment:
I guess, I'm not the right person to judge here because the connector codebase is not really an area where I have much knowledge. But the test you added doesn't cover the fix (neither does reverting the fix of this PR make any of the newly added tests fail nor does any of the tests call the method in question `FileSystemTableSink#toStagingPath`).
That is where I'm wondering what the value of the added tests are. Or am I missing something here? :thinking:
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java:
##########
@@ -374,7 +374,10 @@ public DynamicTableSource.DataStructureConverter createDataStructureConverter(
}
private Path toStagingPath() {
- Path stagingDir = new Path(path, ".staging_" + System.currentTimeMillis());
+ // Add a random UUID to prevent multiple sinks from sharing the same staging dir.
+ // Please see FLINK-29114 for more details
+ Path stagingDir =
+ new Path(path, ".staging_" + UUID.randomUUID() + System.currentTimeMillis());
Review Comment:
```suggestion
new Path(path, ".staging_" + UUID.randomUUID());
```
Maybe, you missed [my previous comment](https://github.com/apache/flink/pull/24390/files#r1503917055) but shouldn't we be able to only use the UUID here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1505814186
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java:
##########
@@ -374,7 +374,10 @@ public DynamicTableSource.DataStructureConverter createDataStructureConverter(
}
private Path toStagingPath() {
- Path stagingDir = new Path(path, ".staging_" + System.currentTimeMillis());
+ // Add a random UUID to prevent multiple sinks from sharing the same staging dir.
+ // Please see FLINK-29114 for more details
+ Path stagingDir =
+ new Path(path, ".staging_" + UUID.randomUUID() + System.currentTimeMillis());
Review Comment:
I intend to keep it to ease troubleshooting. I'm okay with removing it since it's redundant anyway.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1505832996
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
Review Comment:
> [...] the issue is triggered when the FileSystemOutputFormat#finalizeGlobal method is called (triggered by the final block).
I cannot confirm that. The change code segment isn't covered by the newly added tests. As said, reverting your fix in `FileSystemTableSink` doesn't make any of the tests fail.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1505763661
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -104,15 +110,139 @@ void testNonPartition() throws Exception {
.containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
}
+ @Test
+ void testMultiSinkWriteToSameOutputPathUsingSameStagingDir() throws Exception {
+ // sink1
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef1 = new AtomicReference<>();
+ AtomicReference<File> fileToCommitRef1 = new AtomicReference<>();
+ AtomicReference<String> contentRef1 = new AtomicReference<>();
+ try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
+ createSink(false, false, false, new LinkedHashMap<>(), sinkRef1)) {
+ writeUnorderedRecords(testHarness);
+ Map<File, String> content = getFileContentByPath(tmpPath);
+ assertThat(content).hasSize(1);
+ fileToCommitRef1.set(new ArrayList<>(content.keySet()).get(0));
+ contentRef1.set(new ArrayList<>(content.values()).get(0));
+ }
+
+ // sink2
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef2 = new AtomicReference<>();
+ AtomicReference<File> fileToCommitRef2 = new AtomicReference<>();
+ AtomicReference<String> contentRef2 = new AtomicReference<>();
+ try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
+ createSink(false, false, false, new LinkedHashMap<>(), sinkRef2)) {
+ // write different content
+ List<Row> records =
+ Arrays.asList(
+ Row.of("b1", 1, "x1"), Row.of("b2", 2, "x2"), Row.of("b3", 3, "x3"));
+ writeUnorderedRecords(testHarness, records);
+ Map<File, String> content = getFileContentByPath(tmpPath);
+ assertThat(content).hasSize(1);
+ fileToCommitRef2.set(new ArrayList<>(content.keySet()).get(0));
+ contentRef2.set(new ArrayList<>(content.values()).get(0));
+ }
+
+ File fileToCommit1 = fileToCommitRef1.get();
+ File fileToCommit2 = fileToCommitRef2.get();
+ // because sink1 and sink2 are writing to the same staging dir
+ // the file generated by sink1 has been overwritten by sink2
+ assertThat(fileToCommit1.getParent()).isEqualTo(fileToCommit2.getParent());
+ assertThat(fileToCommit1).doesNotExist();
Review Comment:
```suggestion
assertThat(fileToCommit1).hasParent(fileToCommit2.getParent()).doesNotExist();
```
could be merged and simlified a bit
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -104,15 +110,139 @@ void testNonPartition() throws Exception {
.containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
}
+ @Test
+ void testMultiSinkWriteToSameOutputPathUsingSameStagingDir() throws Exception {
+ // sink1
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef1 = new AtomicReference<>();
+ AtomicReference<File> fileToCommitRef1 = new AtomicReference<>();
+ AtomicReference<String> contentRef1 = new AtomicReference<>();
+ try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
+ createSink(false, false, false, new LinkedHashMap<>(), sinkRef1)) {
+ writeUnorderedRecords(testHarness);
+ Map<File, String> content = getFileContentByPath(tmpPath);
+ assertThat(content).hasSize(1);
+ fileToCommitRef1.set(new ArrayList<>(content.keySet()).get(0));
+ contentRef1.set(new ArrayList<>(content.values()).get(0));
+ }
+
+ // sink2
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef2 = new AtomicReference<>();
+ AtomicReference<File> fileToCommitRef2 = new AtomicReference<>();
+ AtomicReference<String> contentRef2 = new AtomicReference<>();
+ try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
+ createSink(false, false, false, new LinkedHashMap<>(), sinkRef2)) {
+ // write different content
+ List<Row> records =
+ Arrays.asList(
+ Row.of("b1", 1, "x1"), Row.of("b2", 2, "x2"), Row.of("b3", 3, "x3"));
+ writeUnorderedRecords(testHarness, records);
+ Map<File, String> content = getFileContentByPath(tmpPath);
+ assertThat(content).hasSize(1);
+ fileToCommitRef2.set(new ArrayList<>(content.keySet()).get(0));
+ contentRef2.set(new ArrayList<>(content.values()).get(0));
+ }
+
+ File fileToCommit1 = fileToCommitRef1.get();
+ File fileToCommit2 = fileToCommitRef2.get();
+ // because sink1 and sink2 are writing to the same staging dir
+ // the file generated by sink1 has been overwritten by sink2
+ assertThat(fileToCommit1.getParent()).isEqualTo(fileToCommit2.getParent());
+ assertThat(fileToCommit1).doesNotExist();
Review Comment:
```suggestion
assertThat(fileToCommit1).hasParent(fileToCommit2.getParent()).doesNotExist();
```
could be merged and simplified a bit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on PR #24390:
URL: https://github.com/apache/flink/pull/24390#issuecomment-1970685919
What do you think if we move `toStagingPath` to `FileSystemOutputFormat`?
Sth. like
```java
private FileSystemOutputFormat(
FileSystemFactory fsFactory,
TableMetaStoreFactory msFactory,
boolean overwrite,
boolean isToLocal,
+ Path path,
- Path tmpPath,
String[] partitionColumns,
boolean dynamicGrouped,
LinkedHashMap<String, String> staticPartitions,
OutputFormatFactory<T> formatFactory,
PartitionComputer<T> computer,
OutputFileConfig outputFileConfig,
ObjectIdentifier identifier,
PartitionCommitPolicyFactory partitionCommitPolicyFactory) {
this.fsFactory = fsFactory;
this.msFactory = msFactory;
this.overwrite = overwrite;
this.isToLocal = isToLocal;
+ this.path = path;
+ this.tmpPath = toStagingPath();
this.partitionColumns = partitionColumns;
this.dynamicGrouped = dynamicGrouped;
this.staticPartitions = staticPartitions;
this.formatFactory = formatFactory;
this.computer = computer;
this.outputFileConfig = outputFileConfig;
this.identifier = identifier;
this.partitionCommitPolicyFactory = partitionCommitPolicyFactory;
}
private Path toStagingPath() {
// Add a random UUID to prevent multiple sinks from sharing the same staging dir.
// Please see FLINK-29114 for more details
Path stagingDir =
new Path(
path,
String.join(
"_",
".staging_",
String.valueOf(System.currentTimeMillis()),
UUID.randomUUID().toString()));
try {
FileSystem fs = stagingDir.getFileSystem();
Preconditions.checkState(
fs.mkdirs(stagingDir), "Failed to create staging dir " + stagingDir);
return stagingDir;
} catch (IOException e) {
throw new RuntimeException(e);
}
}
@VisibleForTesting
public Path getTmpPath() {
return tmpPath;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][table-planner][filesystem][test] Fix unstable TableSourceITCase#testTableHintWithLogicalTableScanReuse [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1503780455
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java:
##########
@@ -374,7 +374,8 @@ public DynamicTableSource.DataStructureConverter createDataStructureConverter(
}
private Path toStagingPath() {
- Path stagingDir = new Path(path, ".staging_" + System.currentTimeMillis());
+ Path stagingDir =
+ new Path(path, ".staging_" + UUID.randomUUID() + System.currentTimeMillis());
Review Comment:
nit: do you thinkt it is worth to add a comment why we need it?
May be not the full description here, however e.g. a link to the corresonding comment in jira?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1521248255
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -282,7 +278,7 @@ private FileSystemOutputFormat<Row> createSinkFormat(
TableMetaStoreFactory msFactory = new FileSystemCommitterTest.TestMetaStoreFactory(path);
return new FileSystemOutputFormat.Builder<Row>()
.setMetaStoreFactory(msFactory)
- .setPath(path)
+ .setStagingPath(new Path(stagingPath.toString()))
Review Comment:
You can still stick to `setPath(stagingBasePath)` here and remove the `stagingPath` field. The only thing you have to change is that you have to assert for empty directory rather than deleted directory in `checkWriteAndCommit`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1507150180
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java:
##########
@@ -386,6 +389,11 @@ private Path toStagingPath() {
}
}
+ @VisibleForTesting
Review Comment:
[Looks like](https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=57965&view=logs&j=fc5181b0-e452-5c8f-68de-1097947f6483&t=995c650b-6573-581c-9ce6-7ad4cc038461&l=23032) the architecture test don't allow this. :thinking:
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java:
##########
@@ -374,7 +375,9 @@ public DynamicTableSource.DataStructureConverter createDataStructureConverter(
}
private Path toStagingPath() {
- Path stagingDir = new Path(path, ".staging_" + System.currentTimeMillis());
+ // Add a random UUID to prevent multiple sinks from sharing the same staging dir.
+ // Please see FLINK-29114 for more details
+ Path stagingDir = new Path(path, ".staging_" + getStagingPathPostfix(false));
try {
FileSystem fs = stagingDir.getFileSystem();
Preconditions.checkState(
Review Comment:
The precondition should be updated. Or did I miss something and you don't agree with [what we discussed in the other PR](https://github.com/LadyForest/flink/pull/1#discussion_r1505972482)?
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -104,15 +113,101 @@ void testNonPartition() throws Exception {
.containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testStagingDirBehavior(boolean shareStagingDir) throws Exception {
Review Comment:
Generally speaking, unit tests should be as small as possible to focus on the specific aspect of the class contract (in this case `FileSystemTableSink#toStagingPath` not returning the same folder twice. This test adds quite a bit of extra context here which might make it harder to grasp the intention for code readers. What's your opinion on that?
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java:
##########
@@ -386,6 +389,11 @@ private Path toStagingPath() {
}
}
+ @VisibleForTesting
+ static String getStagingPathPostfix(boolean constant) {
Review Comment:
You introduce the parameter `constant` to make the test parameterizable. But ideally, you don't want to make production code "to please test code". I feel like this `constant` parameter is not necessary (because the production code only uses `false`). The decision whether you return a fixed suffix (or if you follow the comment above to return the actual path, instead: a fixed path) or the random suffix should live in the test code. WDYT?
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java:
##########
@@ -374,7 +375,9 @@ public DynamicTableSource.DataStructureConverter createDataStructureConverter(
}
private Path toStagingPath() {
- Path stagingDir = new Path(path, ".staging_" + System.currentTimeMillis());
+ // Add a random UUID to prevent multiple sinks from sharing the same staging dir.
+ // Please see FLINK-29114 for more details
+ Path stagingDir = new Path(path, ".staging_" + getStagingPathPostfix(false));
Review Comment:
Why did you decide on creating the `getStagingPathPostfix` method? This creates unnecessary code in the test because we still have to generate the Path there. Alternatively, you could move the entire content of `toStagingPath()` into a static method `generateStagingPath(Path parent, boolean constant)`. That way you just need to call the static method in the test without any path resolution.
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -104,15 +113,101 @@ void testNonPartition() throws Exception {
.containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testStagingDirBehavior(boolean shareStagingDir) throws Exception {
+ // sink1
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef1 = new AtomicReference<>();
+ AtomicReference<Map<File, String>> fileToCommitRef1 = new AtomicReference<>();
+ writeRowsToSink(
+ sinkRef1,
+ fileToCommitRef1,
+ getStagingDir(shareStagingDir),
+ Row.of("a1", 1, "x1"),
+ Row.of("a2", 2, "x2"));
+
+ // sink2
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef2 = new AtomicReference<>();
+ AtomicReference<Map<File, String>> fileToCommitRef2 = new AtomicReference<>();
+ writeRowsToSink(
+ sinkRef2,
+ fileToCommitRef2,
+ getStagingDir(shareStagingDir),
+ Row.of("b1", 1, "y1"),
+ Row.of("b2", 2, "y2"));
+
+ assertSinkBehavior(sinkRef1, fileToCommitRef1, sinkRef2, fileToCommitRef2, shareStagingDir);
+ }
+
+ private void writeRowsToSink(
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef,
+ AtomicReference<Map<File, String>> contentRef,
+ Path stagingDir,
+ Row... records)
+ throws Exception {
+ try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
+ createSink(false, false, false, stagingDir, new LinkedHashMap<>(), sinkRef)) {
+ writeUnorderedRecords(testHarness, Arrays.asList(records));
+ contentRef.set(getFileContentByPath(Paths.get(stagingDir.getPath())));
+ }
+ }
+
+ private Path getStagingDir(boolean shareStagingDir) {
+ String pathPostfix = FileSystemTableSink.getStagingPathPostfix(shareStagingDir);
+ return Path.fromLocalFile(tmpPath.resolve(pathPostfix).toFile());
+ }
+
+ private void assertSinkBehavior(
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef1,
+ AtomicReference<Map<File, String>> fileToCommitRef1,
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef2,
+ AtomicReference<Map<File, String>> fileToCommitRef2,
+ boolean shareStagingDir)
+ throws Exception {
+ Map<File, String> fileToCommit1 = fileToCommitRef1.get();
+ Map<File, String> fileToCommit2 = fileToCommitRef2.get();
+ assertThat(fileToCommit2.keySet()).allMatch(File::exists);
+ if (shareStagingDir) {
+ assertThat(fileToCommit1.keySet()).noneMatch(File::exists);
+ } else {
+ assertThat(fileToCommit1.keySet()).allMatch(File::exists);
+ }
+ sinkRef1.get().finalizeGlobal(finalizationContext);
+ sinkRef2.get().finalizeGlobal(finalizationContext);
+ Map<File, String> committedFiles = getFileContentByPath(outputPath);
+ if (shareStagingDir) {
Review Comment:
I still don't get this test case? Are there scenarios where this is actually desired? Isn't that actually what we want to avoid? And essentially we're fixing it in this PR to make this scenario never appear? :thinking:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1507172607
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java:
##########
@@ -374,7 +375,9 @@ public DynamicTableSource.DataStructureConverter createDataStructureConverter(
}
private Path toStagingPath() {
- Path stagingDir = new Path(path, ".staging_" + System.currentTimeMillis());
+ // Add a random UUID to prevent multiple sinks from sharing the same staging dir.
+ // Please see FLINK-29114 for more details
+ Path stagingDir = new Path(path, ".staging_" + getStagingPathPostfix(false));
try {
FileSystem fs = stagingDir.getFileSystem();
Preconditions.checkState(
Review Comment:
Sorry, I just noticed that message. Agreed with you.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1507204553
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java:
##########
@@ -386,6 +389,11 @@ private Path toStagingPath() {
}
}
+ @VisibleForTesting
Review Comment:
Yes, just noticed that rule...Let me think about it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1521155291
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemOutputFormat.java:
##########
@@ -96,6 +99,22 @@ private FileSystemOutputFormat(
this.outputFileConfig = outputFileConfig;
this.identifier = identifier;
this.partitionCommitPolicyFactory = partitionCommitPolicyFactory;
+
+ createStagingDirectory(this.stagingPath);
+ }
+
+ private static void createStagingDirectory(Path stagingPath) {
+ try {
+ final FileSystem stagingFileSystem = stagingPath.getFileSystem();
+ Preconditions.checkState(
+ !stagingFileSystem.exists(stagingPath),
+ "Staging dir %s already exists",
+ stagingFileSystem);
Review Comment:
Nit: just noticed a typo that it should be `stagingPath` instead of `stagingFileSystem` at L#112
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1511414346
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/utils/PathUtils.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.file.table.utils;
+
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.UUID;
+
+/** Path utils for file system. */
+public class PathUtils {
+
+ public static Path getStagingPath(Path path) {
+ // Add a random UUID to prevent multiple sinks from sharing the same staging dir.
+ // Please see FLINK-29114 for more details
+ Path stagingDir =
+ new Path(
+ path,
+ String.join(
Review Comment:
I'm not sure whether that's actually what you want. The resulting String would be something like `.staging__1709568597254_2cf73aab-39a4-440a-b81c-216be9635bb8` (i.e. double underscore).
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/utils/PathUtils.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.file.table.utils;
+
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.UUID;
+
+/** Path utils for file system. */
+public class PathUtils {
+
+ public static Path getStagingPath(Path path) {
Review Comment:
```suggestion
public static Path getStagingPath(Path path) {
return getStagingPath(
path,
() ->
String.format(
".staging_%d_%s", System.currentTimeMillis(), UUID.randomUUID()));
}
@VisibleForTesting
static Path getStagingPath(Path path, Supplier<String> suffixSupplier) {
```
The current test is "theoretically" flaky. We change that by using a callback and checking for the Precondition. That would enable us to create the following test:
```java
@Test
void testReusingStagingDirFails(@TempDir Path tmpDir) throws IOException {
final String subfolderName = "directory-name";
Files.createDirectory(tmpDir.resolve(subfolderName));
assertThatThrownBy(
() ->
PathUtils.getStagingPath(
org.apache.flink.core.fs.Path.fromLocalFile(
tmpDir.toFile()),
() -> subfolderName))
.isEqualTo(IllegalStateException.class);
}
```
And that would reveal that our assumption was actually wrong that `mkdirs` returns `false` if the directory already exists. :-/
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/utils/PathUtilsTest.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.file.table.utils;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.file.table.FileSystemConnectorOptions;
+import org.apache.flink.core.fs.Path;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+
+/** Test for {@link PathUtils}. */
+public class PathUtilsTest {
+
+ @Test
+ void testUniqueStagingDirectory(@TempDir File tmpDir) {
+ final Configuration config = new Configuration();
+ config.set(FileSystemConnectorOptions.PATH, tmpDir.getAbsolutePath());
+ assertThat(tmpDir.listFiles()).isEmpty();
Review Comment:
```suggestion
```
we wouldn't need this bit anymore, would we?
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/utils/PathUtils.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.file.table.utils;
+
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.UUID;
+
+/** Path utils for file system. */
+public class PathUtils {
+
+ public static Path getStagingPath(Path path) {
+ // Add a random UUID to prevent multiple sinks from sharing the same staging dir.
+ // Please see FLINK-29114 for more details
+ Path stagingDir =
+ new Path(
+ path,
+ String.join(
+ "_",
+ ".staging_",
+ String.valueOf(System.currentTimeMillis()),
+ UUID.randomUUID().toString()));
+ try {
+ FileSystem fs = stagingDir.getFileSystem();
+ Preconditions.checkState(
+ fs.mkdirs(stagingDir), "Failed to create staging dir " + stagingDir);
Review Comment:
```suggestion
Preconditions.checkState(
!fs.exists(stagingDir), "Failed to create staging dir %s", stagingDir);
fs.mkdirs(stagingDir);
```
Looks like that is what we want here. I'm a bit hesitant here because checking for non-existence can be expensive in object storage filesystems like S3 with big buckets. :thinking: But to be fair, it is not a regression because the previous code did the same.
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/utils/PathUtils.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.file.table.utils;
+
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.UUID;
+
+/** Path utils for file system. */
+public class PathUtils {
Review Comment:
`HiveTableSink#createBatchCompactSink` uses the random staging directory for the [BatchFileWriter instantiation in [line 490ff](https://github.com/apache/flink/blob/06caf191991b944769cb7b02769b5d769064febf/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSink.java#L492). The `BatchFileWriter` instantiates a `PartitionTempFileManager` in the operator's open method ([BatchFileWriter:85](https://github.com/apache/flink/blob/0f3470db83c1fddba9ac9a7299b1e61baab4ff12/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/batch/compact/BatchFileWriter.java#L85)). The `PartitionTempFileManager` creates a subfolder for the attempt and deletes any existing directory to ensure that no side effects occur.
That said, the two code locations seem to be independent from each other. Moving the staging directory into the OutputFormat still sounds like a reasonable thing to do. But that's your call in the end.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1505864009
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
Review Comment:
> > [...] the issue is triggered when the FileSystemOutputFormat#finalizeGlobal method is called (triggered by the final block).
>
> I cannot confirm that. The change code segment isn't covered by the newly added tests. As said, reverting your fix in `FileSystemTableSink` doesn't make any of the tests fail.
You can verify the root cause by the following steps.
1. temporarily make the toStagingPath a static method and modify the test case like the following snippet.
```java
@Test
void testMultiSinkWriteToSameOutputPathUsingSameStagingDir() throws Exception {
// sink1
AtomicReference<FileSystemOutputFormat<Row>> sinkRef1 = new AtomicReference<>();
AtomicReference<File> fileToCommitRef1 = new AtomicReference<>();
AtomicReference<String> contentRef1 = new AtomicReference<>();
try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
createSink(
false,
false,
false,
FileSystemTableSink.toStagingPath(Path.fromLocalFile(tmpPath.toFile())),
new LinkedHashMap<>(),
sinkRef1)) {
writeUnorderedRecords(testHarness);
Map<File, String> content = getFileContentByPath(tmpPath);
assertThat(content).hasSize(1);
fileToCommitRef1.set(new ArrayList<>(content.keySet()).get(0));
contentRef1.set(new ArrayList<>(content.values()).get(0));
}
// sink2
AtomicReference<FileSystemOutputFormat<Row>> sinkRef2 = new AtomicReference<>();
AtomicReference<File> fileToCommitRef2 = new AtomicReference<>();
AtomicReference<String> contentRef2 = new AtomicReference<>();
try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
createSink(
false,
false,
false,
FileSystemTableSink.toStagingPath(Path.fromLocalFile(tmpPath.toFile())),
new LinkedHashMap<>(),
sinkRef2)) {
// write different content
List<Row> records =
Arrays.asList(
Row.of("b1", 1, "x1"), Row.of("b2", 2, "x2"), Row.of("b3", 3, "x3"));
writeUnorderedRecords(testHarness, records);
Map<File, String> content = getFileContentByPath(tmpPath);
assertThat(content).hasSize(1);
fileToCommitRef2.set(new ArrayList<>(content.keySet()).get(0));
contentRef2.set(new ArrayList<>(content.values()).get(0));
}
File fileToCommit1 = fileToCommitRef1.get();
File fileToCommit2 = fileToCommitRef2.get();
// because sink1 and sink2 are writing to the same staging dir
// the file generated by sink1 has been overwritten by sink2
assertThat(fileToCommit1.getParent()).isEqualTo(fileToCommit2.getParent());
assertThat(fileToCommit1).doesNotExist();
// lost records
assertThat(contentRef1.get())
.isEqualTo("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
assertThat(fileToCommit2).exists();
assertThat(contentRef2.get()).isEqualTo("b1,1,x1\n" + "b2,2,x2\n" + "b3,3,x3\n");
// let sink1 commit, and it actually commits the file generated by sink2
sinkRef1.get().finalizeGlobal(finalizationContext);
assertThat(getFileContentByPath(outputPath).values()).containsExactly(contentRef2.get());
// sink2 will commit nothing
sinkRef2.get().finalizeGlobal(finalizationContext);
assertThat(getFileContentByPath(outputPath).values()).containsExactly(contentRef2.get());
}
```
If you run this case, it will fail because the staging dir generated by `FileSystemTableSink.toStagingPath` never returns the same path, and no data loss happens.
2. Then, you change the code to return a constant path and rerun the previous case again; it'll be a success, which indicates one file is overwritten by another.
```java
@VisibleForTesting
public static Path toStagingPath(Path path) {
// return a constant path to reproduce the issue
Path stagingDir =
new Path(path, ".staging_foo");
...
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1505871123
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
Review Comment:
> As said, reverting your fix in `FileSystemTableSink` doesn't make any of the tests fail.
This is because the test does not call any method defined `FileSystemTableSink`. However, it simulates the condition where two sinks share the same stagingDir which is denoted by the variable `tmpPath`.
Do you think it would be a good idea if I made FileSystemTableSink#toStagingPath a static method, added the VisibleForTesting annotation, and called it in FileSystemOutputFormatTest?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1505887655
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java:
##########
@@ -374,7 +374,10 @@ public DynamicTableSource.DataStructureConverter createDataStructureConverter(
}
private Path toStagingPath() {
- Path stagingDir = new Path(path, ".staging_" + System.currentTimeMillis());
+ // Add a random UUID to prevent multiple sinks from sharing the same staging dir.
+ // Please see FLINK-29114 for more details
+ Path stagingDir =
+ new Path(path, ".staging_" + UUID.randomUUID() + System.currentTimeMillis());
Review Comment:
Sorry for the misleading. I meant to say "my original intention."
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1505838606
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
Review Comment:
Personally, I believe the crux of the problem is that `FileSystemTableSink#toStagingPath` acts like a remote time bomb, which only detonates within `FileSystemOutputFormat` when multiple sinks write to the same physical table and create the staging directory at the exact same time. Because it spans multiple components, constructing a unit test is quite challenging. Another approach would be to add an ITCase (in fact, that unstable case could, to some extent, be considered an ITCase for this issue).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1512083846
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/utils/PathUtilsTest.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.file.table.utils;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.file.table.FileSystemConnectorOptions;
+import org.apache.flink.core.fs.Path;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+
+/** Test for {@link PathUtils}. */
+public class PathUtilsTest {
+
+ @Test
+ void testUniqueStagingDirectory(@TempDir File tmpDir) {
+ final Configuration config = new Configuration();
+ config.set(FileSystemConnectorOptions.PATH, tmpDir.getAbsolutePath());
+ assertThat(tmpDir.listFiles()).isEmpty();
Review Comment:
You're right. We don't need this anymore.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1512374782
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/utils/PathUtils.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.file.table.utils;
+
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.UUID;
+
+/** Path utils for file system. */
+public class PathUtils {
+
+ public static Path getStagingPath(Path path) {
Review Comment:
1. Generate duplicate code for temporary file creation in `FileSystemOutputFormat` and `HiveTableSink`. (similar to the base code version of this PR) which should be ok because it's not really complex code that we're dealing with, here.
2. Move the folder generation into a dedicated method that needs to be annotated as `@Public` to honor the ArchUnit tests. That would be a viable solution if we think that the temporary folder is in some means relevant to the user (e.g. is the naming scheme for the staging directory relevant so that it should be not changed between releases?). In that case, we should, indeed, annotate it as `@Public`.
To me, it sounds like the staging directory generation is an internal matter. How important is it that the naming scheme is matched between the `FileSystemTableSink` and the `HiveTableSink`? I would lean towards making the temp file generation internal functionality that is independent between `FileSystemTableSink` and `HiveTableSink` (option 1). But I'm not familiar with this code to judge here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1520757091
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -70,6 +85,11 @@ private static Map<File, String> getFileContentByPath(java.nio.file.Path directo
return contents;
}
+ private static Map<File, String> getStagingFileContent(
Review Comment:
> What we want to make sure is that the class actually creates its own staging folder
I believe this is the point we want to verify after moving the generation of the staging folder to `FileSystemOutputFormat`.
I understand your concern. In fact, `PartitionWriterTest` and `FileSystemCommitterTest` have already covered the checks for written content. Therefore, the tests for `FileSystemOutputFormat` should indeed focus more on the creation/deletion of directories. If we were to check file content as well, it would resemble an IT case.
I guess the reason content is checked is that the test author believes `FileSystemOutputFormat` has the capability to write records (implemented through the `writeRecord` method) and commit files from the staging folder to the outputPath (achieved with `finalizeGlobal`). Thus, it tested the written content even before our modifications were made.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest merged PR #24390:
URL: https://github.com/apache/flink/pull/24390
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1505814697
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java:
##########
@@ -374,7 +374,10 @@ public DynamicTableSource.DataStructureConverter createDataStructureConverter(
}
private Path toStagingPath() {
- Path stagingDir = new Path(path, ".staging_" + System.currentTimeMillis());
+ // Add a random UUID to prevent multiple sinks from sharing the same staging dir.
+ // Please see FLINK-29114 for more details
+ Path stagingDir =
+ new Path(path, ".staging_" + UUID.randomUUID() + System.currentTimeMillis());
Review Comment:
Sorry, I missed the comment. I intend to keep it to ease troubleshooting. I'm okay with removing it since it's redundant anyway.
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java:
##########
@@ -374,7 +374,10 @@ public DynamicTableSource.DataStructureConverter createDataStructureConverter(
}
private Path toStagingPath() {
- Path stagingDir = new Path(path, ".staging_" + System.currentTimeMillis());
+ // Add a random UUID to prevent multiple sinks from sharing the same staging dir.
+ // Please see FLINK-29114 for more details
+ Path stagingDir =
+ new Path(path, ".staging_" + UUID.randomUUID() + System.currentTimeMillis());
Review Comment:
I intend to keep it to ease troubleshooting. I'm okay with removing it since it's redundant anyway.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][table-planner][filesystem][test] Fix unstable TableSourceITCase#testTableHintWithLogicalTableScanReuse [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #24390:
URL: https://github.com/apache/flink/pull/24390#issuecomment-1965970037
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ece0f829708723d1a0fedda0b24c20e5f709e116",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ece0f829708723d1a0fedda0b24c20e5f709e116",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ece0f829708723d1a0fedda0b24c20e5f709e116 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1507288792
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -104,15 +113,101 @@ void testNonPartition() throws Exception {
.containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testStagingDirBehavior(boolean shareStagingDir) throws Exception {
Review Comment:
I guess we're getting closer here. :-) I'm not that familiar with this part of the code. But as said earlier, a complicated verbose unit test is an indication for a non-optimal code structure. Also checking the interfaces' JavaDocs backs your proposal:
For `DynamicTableSink` (which is what `FileSystemTableSink` implements):
> Sink of a dynamic table to an external storage system.
For `OutputFormat` (which is what `FileSystemOutputFormat` implements):
> The base interface for outputs that consumes records. The output format describes how to store the final records, for example in a file.
The key sentence here is `The output format describes how to store the final records, for example in a file.`. The "how" should include the "where", i.e. the `OutputFormat` implementation should be in charge of creating the staging folder and, therefore, would be also in charge of create a folder that is not used by any other `OutputFormat` instance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1507274806
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -104,15 +113,101 @@ void testNonPartition() throws Exception {
.containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testStagingDirBehavior(boolean shareStagingDir) throws Exception {
+ // sink1
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef1 = new AtomicReference<>();
+ AtomicReference<Map<File, String>> fileToCommitRef1 = new AtomicReference<>();
+ writeRowsToSink(
+ sinkRef1,
+ fileToCommitRef1,
+ getStagingDir(shareStagingDir),
+ Row.of("a1", 1, "x1"),
+ Row.of("a2", 2, "x2"));
+
+ // sink2
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef2 = new AtomicReference<>();
+ AtomicReference<Map<File, String>> fileToCommitRef2 = new AtomicReference<>();
+ writeRowsToSink(
+ sinkRef2,
+ fileToCommitRef2,
+ getStagingDir(shareStagingDir),
+ Row.of("b1", 1, "y1"),
+ Row.of("b2", 2, "y2"));
+
+ assertSinkBehavior(sinkRef1, fileToCommitRef1, sinkRef2, fileToCommitRef2, shareStagingDir);
+ }
+
+ private void writeRowsToSink(
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef,
+ AtomicReference<Map<File, String>> contentRef,
+ Path stagingDir,
+ Row... records)
+ throws Exception {
+ try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
+ createSink(false, false, false, stagingDir, new LinkedHashMap<>(), sinkRef)) {
+ writeUnorderedRecords(testHarness, Arrays.asList(records));
+ contentRef.set(getFileContentByPath(Paths.get(stagingDir.getPath())));
+ }
+ }
+
+ private Path getStagingDir(boolean shareStagingDir) {
+ String pathPostfix = FileSystemTableSink.getStagingPathPostfix(shareStagingDir);
+ return Path.fromLocalFile(tmpPath.resolve(pathPostfix).toFile());
+ }
+
+ private void assertSinkBehavior(
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef1,
+ AtomicReference<Map<File, String>> fileToCommitRef1,
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef2,
+ AtomicReference<Map<File, String>> fileToCommitRef2,
+ boolean shareStagingDir)
+ throws Exception {
+ Map<File, String> fileToCommit1 = fileToCommitRef1.get();
+ Map<File, String> fileToCommit2 = fileToCommitRef2.get();
+ assertThat(fileToCommit2.keySet()).allMatch(File::exists);
+ if (shareStagingDir) {
+ assertThat(fileToCommit1.keySet()).noneMatch(File::exists);
+ } else {
+ assertThat(fileToCommit1.keySet()).allMatch(File::exists);
+ }
+ sinkRef1.get().finalizeGlobal(finalizationContext);
+ sinkRef2.get().finalizeGlobal(finalizationContext);
+ Map<File, String> committedFiles = getFileContentByPath(outputPath);
+ if (shareStagingDir) {
Review Comment:
You're understanding is correct. It's just that this test is actually testing `FileSystemOutputFormat` for a scenario where some other class (in our case `FileSystemTableSink`) has a bug.
This test is fine if you think that `FileSystemOutputFormat` should be able to handle this case. But I'm wondering whether it would be more appropriate to fail in such a case where multiple instances are accessing the same folder (but there's the question how easily a `FileSystemOutputFormat` instance can detect something like that.
Anyway, resolving this test scenario easily indicates that the scope of the different classes is not well-defined.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1520696809
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -83,153 +103,175 @@ void after() {
@Test
void testClosingWithoutInput() throws Exception {
try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, false, false, new LinkedHashMap<>(), new AtomicReference<>())) {
+ createTestHarness(
+ false, false, false, new LinkedHashMap<>(), new AtomicReference<>())) {
testHarness.setup();
testHarness.open();
}
}
@Test
void testNonPartition() throws Exception {
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, false, false, new LinkedHashMap<>(), ref)) {
- writeUnorderedRecords(testHarness);
- assertThat(getFileContentByPath(tmpPath)).hasSize(1);
- }
-
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- assertThat(content.values())
- .containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
- }
-
- private void writeUnorderedRecords(OneInputStreamOperatorTestHarness<Row, Object> testHarness)
- throws Exception {
- testHarness.setup();
- testHarness.open();
-
- testHarness.processElement(new StreamRecord<>(Row.of("a1", 1, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p2"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a3", 3, "p1"), 1L));
+ checkWriteAndCommit(
+ false,
+ false,
+ false,
+ new LinkedHashMap<>(),
+ DEFAULT_INPUT_SUPPLIER,
+ DEFAULT_OUTPUT_SUPPLIER);
}
@Test
void testOverrideNonPartition() throws Exception {
testNonPartition();
-
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(true, false, false, new LinkedHashMap<>(), ref)) {
- writeUnorderedRecords(testHarness);
- assertThat(getFileContentByPath(tmpPath)).hasSize(1);
- }
-
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- assertThat(content).hasSize(1);
- assertThat(content.values())
- .containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
- assertThat(new File(tmpPath.toUri())).doesNotExist();
+ checkWriteAndCommit(
+ true,
+ false,
+ false,
+ new LinkedHashMap<>(),
+ DEFAULT_INPUT_SUPPLIER,
+ DEFAULT_OUTPUT_SUPPLIER);
}
@Test
void testStaticPartition() throws Exception {
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
LinkedHashMap<String, String> staticParts = new LinkedHashMap<>();
staticParts.put("c", "p1");
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, true, false, staticParts, ref)) {
- testHarness.setup();
- testHarness.open();
-
- testHarness.processElement(new StreamRecord<>(Row.of("a1", 1), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a3", 3), 1L));
- assertThat(getFileContentByPath(tmpPath)).hasSize(1);
- }
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- assertThat(content).hasSize(1);
- assertThat(content.keySet().iterator().next().getParentFile().getName()).isEqualTo("c=p1");
- assertThat(content.values()).containsExactly("a1,1\n" + "a2,2\n" + "a2,2\n" + "a3,3\n");
- assertThat(new File(tmpPath.toUri())).doesNotExist();
+ checkWriteAndCommit(
+ false,
+ true,
+ false,
+ staticParts,
+ () ->
+ Arrays.asList(
+ new StreamRecord<>(Row.of("a1", 1), 1L),
+ new StreamRecord<>(Row.of("a2", 2), 1L),
+ new StreamRecord<>(Row.of("a2", 2), 1L),
+ new StreamRecord<>(Row.of("a3", 3), 1L)),
+ () ->
+ Collections.singletonList(
+ "c=p1:" + "a1,1\n" + "a2,2\n" + "a2,2\n" + "a3,3\n"));
}
@Test
void testDynamicPartition() throws Exception {
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, true, false, new LinkedHashMap<>(), ref)) {
- writeUnorderedRecords(testHarness);
- assertThat(getFileContentByPath(tmpPath)).hasSize(2);
- }
-
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- Map<String, String> sortedContent = new TreeMap<>();
- content.forEach((file, s) -> sortedContent.put(file.getParentFile().getName(), s));
-
- assertThat(sortedContent).hasSize(2);
- assertThat(sortedContent)
- .contains(entry("c=p1", "a1,1\n" + "a2,2\n" + "a3,3\n"), entry("c=p2", "a2,2\n"));
- assertThat(new File(tmpPath.toUri())).doesNotExist();
+ checkWriteAndCommit(
+ false,
+ true,
+ false,
+ new LinkedHashMap<>(),
+ DEFAULT_INPUT_SUPPLIER,
+ () -> Arrays.asList("c=p1:" + "a1,1\n" + "a2,2\n" + "a3,3\n", "c=p2:" + "a2,2\n"));
}
@Test
void testGroupedDynamicPartition() throws Exception {
+ checkWriteAndCommit(
+ false,
+ true,
+ true,
+ new LinkedHashMap<>(),
+ () ->
+ Arrays.asList(
+ new StreamRecord<>(Row.of("a1", 1, "p1"), 1L),
+ new StreamRecord<>(Row.of("a2", 2, "p1"), 1L),
+ new StreamRecord<>(Row.of("a3", 3, "p1"), 1L),
+ new StreamRecord<>(Row.of("a2", 2, "p2"), 1L)),
+ () -> Arrays.asList("c=p1:" + "a1,1\n" + "a2,2\n" + "a3,3\n", "c=p2:" + "a2,2\n"));
+ }
+
+ @Test
+ void testGetUniqueStagingDirectory() {
+ final int expectedStagingDirectories = 100;
+ for (int i = 0; i < expectedStagingDirectories; i++) {
+ assertDoesNotThrow(() -> createSinkFormat(false, false, false, new LinkedHashMap<>()));
+ }
+ assertThat(outputPath.toFile().listFiles()).hasSize(expectedStagingDirectories);
+ }
+
+ private void checkWriteAndCommit(
+ boolean override,
+ boolean partitioned,
+ boolean dynamicGrouped,
+ LinkedHashMap<String, String> staticPartitions,
+ Supplier<List<StreamRecord<Row>>> inputSupplier,
+ Supplier<List<String>> outputSupplier)
+ throws Exception {
+ List<String> expectedOutput = outputSupplier.get();
+ int expectedFileNum = expectedOutput.size();
AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, true, true, new LinkedHashMap<>(), ref)) {
+ createTestHarness(override, partitioned, dynamicGrouped, staticPartitions, ref)) {
testHarness.setup();
testHarness.open();
-
- testHarness.processElement(new StreamRecord<>(Row.of("a1", 1, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a3", 3, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p2"), 1L));
- assertThat(getFileContentByPath(tmpPath)).hasSize(2);
+ for (StreamRecord<Row> record : inputSupplier.get()) {
+ testHarness.processElement(record);
+ }
+ assertThat(getStagingFileContent(ref)).hasSize(expectedFileNum);
}
ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- Map<String, String> sortedContent = new TreeMap<>();
- content.forEach((file, s) -> sortedContent.put(file.getParentFile().getName(), s));
+ assertThat(Paths.get(ref.get().getStagingPath().getPath())).doesNotExist();
+
+ Map<File, String> fileToContent = getFileContentByPath(outputPath);
+ assertThat(fileToContent).hasSize(expectedFileNum);
+ if (partitioned) {
Review Comment:
For non-partitioned tables, the output file is directly under the outputPath, like `outputPath/part-xxx`. For partitioned tables, each partition will have its own subdirectory, such as `outputPath/year=2024/part-xxx`. I understand that the purpose of this distinction is to ensure that when writing to multiple partitions at the same time, the data for each partition is as expected during testing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1521036729
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -83,157 +110,193 @@ void after() {
@Test
void testClosingWithoutInput() throws Exception {
try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, false, false, new LinkedHashMap<>(), new AtomicReference<>())) {
+ createTestHarness(createSinkFormat(false, false, false, new LinkedHashMap<>()))) {
testHarness.setup();
testHarness.open();
}
}
@Test
void testNonPartition() throws Exception {
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, false, false, new LinkedHashMap<>(), ref)) {
- writeUnorderedRecords(testHarness);
- assertThat(getFileContentByPath(tmpPath)).hasSize(1);
- }
-
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- assertThat(content.values())
- .containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
- }
-
- private void writeUnorderedRecords(OneInputStreamOperatorTestHarness<Row, Object> testHarness)
- throws Exception {
- testHarness.setup();
- testHarness.open();
-
- testHarness.processElement(new StreamRecord<>(Row.of("a1", 1, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p2"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a3", 3, "p1"), 1L));
+ checkWriteAndCommit(
+ false,
+ false,
+ false,
+ new LinkedHashMap<>(),
+ DEFAULT_INPUT_SUPPLIER,
+ DEFAULT_OUTPUT_SUPPLIER);
}
@Test
void testOverrideNonPartition() throws Exception {
testNonPartition();
-
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(true, false, false, new LinkedHashMap<>(), ref)) {
- writeUnorderedRecords(testHarness);
- assertThat(getFileContentByPath(tmpPath)).hasSize(1);
- }
-
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- assertThat(content).hasSize(1);
- assertThat(content.values())
- .containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
- assertThat(new File(tmpPath.toUri())).doesNotExist();
+ checkWriteAndCommit(
+ true,
+ false,
+ false,
+ new LinkedHashMap<>(),
+ DEFAULT_INPUT_SUPPLIER,
+ DEFAULT_OUTPUT_SUPPLIER);
}
@Test
void testStaticPartition() throws Exception {
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
LinkedHashMap<String, String> staticParts = new LinkedHashMap<>();
staticParts.put("c", "p1");
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, true, false, staticParts, ref)) {
- testHarness.setup();
- testHarness.open();
-
- testHarness.processElement(new StreamRecord<>(Row.of("a1", 1), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a3", 3), 1L));
- assertThat(getFileContentByPath(tmpPath)).hasSize(1);
- }
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- assertThat(content).hasSize(1);
- assertThat(content.keySet().iterator().next().getParentFile().getName()).isEqualTo("c=p1");
- assertThat(content.values()).containsExactly("a1,1\n" + "a2,2\n" + "a2,2\n" + "a3,3\n");
- assertThat(new File(tmpPath.toUri())).doesNotExist();
+ checkWriteAndCommit(
+ false,
+ true,
+ false,
+ staticParts,
+ () ->
+ Arrays.asList(
+ new StreamRecord<>(Row.of("a1", 1), 1L),
+ new StreamRecord<>(Row.of("a2", 2), 1L),
+ new StreamRecord<>(Row.of("a2", 2), 1L),
+ new StreamRecord<>(Row.of("a3", 3), 1L)),
+ () ->
+ Collections.singletonMap(
+ "c=p1", createFileContent("a1,1", "a2,2", "a2,2", "a3,3")));
}
@Test
void testDynamicPartition() throws Exception {
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, true, false, new LinkedHashMap<>(), ref)) {
- writeUnorderedRecords(testHarness);
- assertThat(getFileContentByPath(tmpPath)).hasSize(2);
- }
-
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- Map<String, String> sortedContent = new TreeMap<>();
- content.forEach((file, s) -> sortedContent.put(file.getParentFile().getName(), s));
-
- assertThat(sortedContent).hasSize(2);
- assertThat(sortedContent)
- .contains(entry("c=p1", "a1,1\n" + "a2,2\n" + "a3,3\n"), entry("c=p2", "a2,2\n"));
- assertThat(new File(tmpPath.toUri())).doesNotExist();
+ checkWriteAndCommit(
+ false,
+ true,
+ false,
+ new LinkedHashMap<>(),
+ DEFAULT_INPUT_SUPPLIER,
+ () ->
+ ImmutableMap.of(
+ "c=p1",
+ createFileContent("a1,1", "a2,2", "a3,3"),
+ "c=p2",
+ createFileContent("a2,2")));
}
@Test
void testGroupedDynamicPartition() throws Exception {
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
+ checkWriteAndCommit(
+ false,
+ true,
+ true,
+ new LinkedHashMap<>(),
+ () ->
+ Arrays.asList(
+ new StreamRecord<>(Row.of("a1", 1, "p1"), 1L),
+ new StreamRecord<>(Row.of("a2", 2, "p1"), 1L),
+ new StreamRecord<>(Row.of("a3", 3, "p1"), 1L),
+ new StreamRecord<>(Row.of("a2", 2, "p2"), 1L)),
+ () ->
+ ImmutableMap.of(
+ "c=p1",
+ createFileContent("a1,1", "a2,2", "a3,3"),
+ "c=p2",
+ createFileContent("a2,2")));
+ }
+
+ @Test
+ void testGetUniqueStagingDirectory() throws IOException {
+ final Path alreadyExistingStagingDir = new Path(outputPath.toFile().getAbsolutePath());
+ assertThat(alreadyExistingStagingDir.getFileSystem().exists(alreadyExistingStagingDir))
+ .as("The staging folder should already exist.")
+ .isTrue();
+
+ final FileSystemOutputFormat.Builder<Row> builder =
+ new FileSystemOutputFormat.Builder<Row>()
+ .setPartitionColumns(new String[0])
+ .setFormatFactory(TextOutputFormat::new)
+ .setMetaStoreFactory(
+ new FileSystemCommitterTest.TestMetaStoreFactory(
+ new Path(outputPath.toFile().getAbsolutePath())))
+ .setPartitionComputer(
+ new RowPartitionComputer("default", new String[0], new String[0]))
+ .setStagingPath(alreadyExistingStagingDir);
+
+ assertThatThrownBy(builder::build)
+ .as("Reusing a folder should cause an error.")
+ .isInstanceOf(IllegalStateException.class);
+ }
+
+ @SuppressWarnings("unchecked")
+ private void checkWriteAndCommit(
+ boolean override,
+ boolean partitioned,
+ boolean dynamicGrouped,
+ LinkedHashMap<String, String> staticPartitions,
+ Supplier<List<StreamRecord<Row>>> inputSupplier,
+ Supplier<?> outputSupplier)
+ throws Exception {
+ Object expectedOutput = outputSupplier.get();
+ int expectedFileNum =
+ (partitioned)
+ ? ((Map<String, String>) expectedOutput).size()
+ : ((List<String>) expectedOutput).size();
+ FileSystemOutputFormat<Row> outputFormat =
+ createSinkFormat(override, partitioned, dynamicGrouped, staticPartitions);
try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, true, true, new LinkedHashMap<>(), ref)) {
+ createTestHarness(outputFormat)) {
testHarness.setup();
testHarness.open();
+ for (StreamRecord<Row> record : inputSupplier.get()) {
+ testHarness.processElement(record);
+ }
+ assertThat(getStagingFileContent(outputFormat)).hasSize(expectedFileNum);
+ }
+
+ outputFormat.finalizeGlobal(finalizationContext);
+ assertThat(Paths.get(outputFormat.getStagingPath().getPath())).doesNotExist();
- testHarness.processElement(new StreamRecord<>(Row.of("a1", 1, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a3", 3, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p2"), 1L));
- assertThat(getFileContentByPath(tmpPath)).hasSize(2);
+ Map<File, String> fileToContent = getFileContentByPath(outputPath);
+ assertThat(fileToContent).hasSize(expectedFileNum);
+ if (partitioned) {
+ assertPartitionedOutput((Map<String, String>) expectedOutput, fileToContent);
+ } else {
+ assertThat(fileToContent.values())
+ .containsExactlyInAnyOrderElementsOf((List<String>) expectedOutput);
}
+ }
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- Map<String, String> sortedContent = new TreeMap<>();
- content.forEach((file, s) -> sortedContent.put(file.getParentFile().getName(), s));
+ private void assertPartitionedOutput(
Review Comment:
nit: that method could be removed and inlined in the `if (partitioned)` block of the calling method to improve readbility.
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -70,6 +85,11 @@ private static Map<File, String> getFileContentByPath(java.nio.file.Path directo
return contents;
}
+ private static Map<File, String> getStagingFileContent(
Review Comment:
hm, the existence and non-existence of files in the staging folder can be verified without revealing the staging directory. I guess, the mistake we made is that we reuse the `outputPath` temporary folder in for the `TestMetaStoreFactory` and the `FileSystemOutputFormat`. If we would use two separate folders (like it was done before our change), we could just check recursively that, indeed, the expected amount of intermediate files is created by the `FileSystemOutputFormat` and that all content is deleted after the `finalizeGlobal` call in `checkWriteAndCommt`.
The reason I am stressing on that part is that I'd like to remove the `FileSystemOutputFormat#getStagingFolder` method. That seems to reveal internal state without a reason. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1521128496
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -70,6 +85,11 @@ private static Map<File, String> getFileContentByPath(java.nio.file.Path directo
return contents;
}
+ private static Map<File, String> getStagingFileContent(
Review Comment:
> The reason I am stressing on that part is that I'd like to remove the FileSystemOutputFormat#getStagingFolder method.
Good point! Let me remove it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1512043134
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/utils/PathUtils.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.file.table.utils;
+
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.UUID;
+
+/** Path utils for file system. */
+public class PathUtils {
+
+ public static Path getStagingPath(Path path) {
+ // Add a random UUID to prevent multiple sinks from sharing the same staging dir.
+ // Please see FLINK-29114 for more details
+ Path stagingDir =
+ new Path(
+ path,
+ String.join(
Review Comment:
Sorry, that's a typo.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1512605257
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/utils/PathUtils.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.file.table.utils;
+
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.UUID;
+
+/** Path utils for file system. */
+public class PathUtils {
+
+ public static Path getStagingPath(Path path) {
Review Comment:
I'm opting for option one as well. I didn't choose it in the first place because
<1> Not very sure if it is tolerable to have duplicate logic scattered throughout the codebase.
<2> I was hesitant about whether altering `FileSystemOutputFormat$Builder#setTempPath` to `FileSystemOutputFormat$Builder#setPath` might lead to any issues with dependencies. While I've manually reviewed the connector repositories and found no instances of the method being called, the impact on custom connectors remains unclear. Perhaps my concerns are a bit excessive.
It appears that we're on the same page. I'll address the comment as we've discussed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1519417563
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -83,153 +103,175 @@ void after() {
@Test
void testClosingWithoutInput() throws Exception {
try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, false, false, new LinkedHashMap<>(), new AtomicReference<>())) {
+ createTestHarness(
+ false, false, false, new LinkedHashMap<>(), new AtomicReference<>())) {
testHarness.setup();
testHarness.open();
}
}
@Test
void testNonPartition() throws Exception {
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, false, false, new LinkedHashMap<>(), ref)) {
- writeUnorderedRecords(testHarness);
- assertThat(getFileContentByPath(tmpPath)).hasSize(1);
- }
-
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- assertThat(content.values())
- .containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
- }
-
- private void writeUnorderedRecords(OneInputStreamOperatorTestHarness<Row, Object> testHarness)
- throws Exception {
- testHarness.setup();
- testHarness.open();
-
- testHarness.processElement(new StreamRecord<>(Row.of("a1", 1, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p2"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a3", 3, "p1"), 1L));
+ checkWriteAndCommit(
+ false,
+ false,
+ false,
+ new LinkedHashMap<>(),
+ DEFAULT_INPUT_SUPPLIER,
+ DEFAULT_OUTPUT_SUPPLIER);
}
@Test
void testOverrideNonPartition() throws Exception {
testNonPartition();
-
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(true, false, false, new LinkedHashMap<>(), ref)) {
- writeUnorderedRecords(testHarness);
- assertThat(getFileContentByPath(tmpPath)).hasSize(1);
- }
-
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- assertThat(content).hasSize(1);
- assertThat(content.values())
- .containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
- assertThat(new File(tmpPath.toUri())).doesNotExist();
+ checkWriteAndCommit(
+ true,
+ false,
+ false,
+ new LinkedHashMap<>(),
+ DEFAULT_INPUT_SUPPLIER,
+ DEFAULT_OUTPUT_SUPPLIER);
}
@Test
void testStaticPartition() throws Exception {
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
LinkedHashMap<String, String> staticParts = new LinkedHashMap<>();
staticParts.put("c", "p1");
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, true, false, staticParts, ref)) {
- testHarness.setup();
- testHarness.open();
-
- testHarness.processElement(new StreamRecord<>(Row.of("a1", 1), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a3", 3), 1L));
- assertThat(getFileContentByPath(tmpPath)).hasSize(1);
- }
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- assertThat(content).hasSize(1);
- assertThat(content.keySet().iterator().next().getParentFile().getName()).isEqualTo("c=p1");
- assertThat(content.values()).containsExactly("a1,1\n" + "a2,2\n" + "a2,2\n" + "a3,3\n");
- assertThat(new File(tmpPath.toUri())).doesNotExist();
+ checkWriteAndCommit(
+ false,
+ true,
+ false,
+ staticParts,
+ () ->
+ Arrays.asList(
+ new StreamRecord<>(Row.of("a1", 1), 1L),
+ new StreamRecord<>(Row.of("a2", 2), 1L),
+ new StreamRecord<>(Row.of("a2", 2), 1L),
+ new StreamRecord<>(Row.of("a3", 3), 1L)),
+ () ->
+ Collections.singletonList(
+ "c=p1:" + "a1,1\n" + "a2,2\n" + "a2,2\n" + "a3,3\n"));
}
@Test
void testDynamicPartition() throws Exception {
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, true, false, new LinkedHashMap<>(), ref)) {
- writeUnorderedRecords(testHarness);
- assertThat(getFileContentByPath(tmpPath)).hasSize(2);
- }
-
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- Map<String, String> sortedContent = new TreeMap<>();
- content.forEach((file, s) -> sortedContent.put(file.getParentFile().getName(), s));
-
- assertThat(sortedContent).hasSize(2);
- assertThat(sortedContent)
- .contains(entry("c=p1", "a1,1\n" + "a2,2\n" + "a3,3\n"), entry("c=p2", "a2,2\n"));
- assertThat(new File(tmpPath.toUri())).doesNotExist();
+ checkWriteAndCommit(
+ false,
+ true,
+ false,
+ new LinkedHashMap<>(),
+ DEFAULT_INPUT_SUPPLIER,
+ () -> Arrays.asList("c=p1:" + "a1,1\n" + "a2,2\n" + "a3,3\n", "c=p2:" + "a2,2\n"));
}
@Test
void testGroupedDynamicPartition() throws Exception {
+ checkWriteAndCommit(
+ false,
+ true,
+ true,
+ new LinkedHashMap<>(),
+ () ->
+ Arrays.asList(
+ new StreamRecord<>(Row.of("a1", 1, "p1"), 1L),
+ new StreamRecord<>(Row.of("a2", 2, "p1"), 1L),
+ new StreamRecord<>(Row.of("a3", 3, "p1"), 1L),
+ new StreamRecord<>(Row.of("a2", 2, "p2"), 1L)),
+ () -> Arrays.asList("c=p1:" + "a1,1\n" + "a2,2\n" + "a3,3\n", "c=p2:" + "a2,2\n"));
+ }
+
+ @Test
+ void testGetUniqueStagingDirectory() {
+ final int expectedStagingDirectories = 100;
+ for (int i = 0; i < expectedStagingDirectories; i++) {
+ assertDoesNotThrow(() -> createSinkFormat(false, false, false, new LinkedHashMap<>()));
+ }
+ assertThat(outputPath.toFile().listFiles()).hasSize(expectedStagingDirectories);
+ }
+
+ private void checkWriteAndCommit(
+ boolean override,
+ boolean partitioned,
+ boolean dynamicGrouped,
+ LinkedHashMap<String, String> staticPartitions,
+ Supplier<List<StreamRecord<Row>>> inputSupplier,
+ Supplier<List<String>> outputSupplier)
+ throws Exception {
+ List<String> expectedOutput = outputSupplier.get();
+ int expectedFileNum = expectedOutput.size();
AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, true, true, new LinkedHashMap<>(), ref)) {
+ createTestHarness(override, partitioned, dynamicGrouped, staticPartitions, ref)) {
testHarness.setup();
testHarness.open();
-
- testHarness.processElement(new StreamRecord<>(Row.of("a1", 1, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a3", 3, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p2"), 1L));
- assertThat(getFileContentByPath(tmpPath)).hasSize(2);
+ for (StreamRecord<Row> record : inputSupplier.get()) {
+ testHarness.processElement(record);
+ }
+ assertThat(getStagingFileContent(ref)).hasSize(expectedFileNum);
}
ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- Map<String, String> sortedContent = new TreeMap<>();
- content.forEach((file, s) -> sortedContent.put(file.getParentFile().getName(), s));
+ assertThat(Paths.get(ref.get().getStagingPath().getPath())).doesNotExist();
+
+ Map<File, String> fileToContent = getFileContentByPath(outputPath);
+ assertThat(fileToContent).hasSize(expectedFileNum);
+ if (partitioned) {
Review Comment:
Why do we handle partitioned output separately? Isn't the non-partitioned output just a special case of the partition output (with a single partition being present)?
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemOutputFormat.java:
##########
@@ -197,13 +200,37 @@ public void close() throws IOException {
}
}
+ private Path toStagingPath(Path path) {
+ // Add a random UUID to prevent multiple sinks from sharing the same staging dir.
+ // Please see FLINK-29114 for more details
Review Comment:
```suggestion
```
nit: I'm not sure whether this comment is still necessary. In `FileSystemOutputFormat`, we have a dedicated folder now that is used for this specific instance and is, therefore, a implementation detail that's hidden from the "outside".
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -37,24 +37,39 @@
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
-import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.entry;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
/** Test for {@link FileSystemOutputFormat}. */
class FileSystemOutputFormatTest {
- @TempDir private java.nio.file.Path tmpPath;
@TempDir private java.nio.file.Path outputPath;
private final TestingFinalizationContext finalizationContext = new TestingFinalizationContext();
+ private static final Supplier<List<StreamRecord<Row>>> DEFAULT_INPUT_SUPPLIER =
+ () ->
+ Arrays.asList(
+ new StreamRecord<>(Row.of("a1", 1, "p1"), 1L),
+ new StreamRecord<>(Row.of("a2", 2, "p1"), 1L),
+ new StreamRecord<>(Row.of("a2", 2, "p2"), 1L),
+ new StreamRecord<>(Row.of("a3", 3, "p1"), 1L));
+
+ private static final Supplier<List<String>> DEFAULT_OUTPUT_SUPPLIER =
+ () -> Collections.singletonList("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
Review Comment:
```suggestion
() ->
Collections.singletonList(
createFileContent("a1,1,p1", "a2,2,p1", "a2,2,p2", "a3,3,p1"));
private static String createFileContent(String... rows) {
return Arrays.stream(rows).collect(Collectors.joining("\n", "", "\n"));
}
```
Looks like we could move the file content generation into its own utility method to hide the newline format since it's used in numerous places within this test. WDYT?
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemOutputFormat.java:
##########
@@ -197,13 +200,37 @@ public void close() throws IOException {
}
}
+ private Path toStagingPath(Path path) {
Review Comment:
```suggestion
private Path createStagingPath(Path path) {
```
nit: more descriptive name
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -70,6 +85,11 @@ private static Map<File, String> getFileContentByPath(java.nio.file.Path directo
return contents;
}
+ private static Map<File, String> getStagingFileContent(
Review Comment:
I'm wondering whether we actually need to check the staging content here. It's an implementation detail how the data is stored in the folder. What we want to make sure is that the class actually creates its own staging folder (rather than relying on the folder that is passed in through the constructor).
That would make most of the refactoring of this test class obsolete, I guess. :thinking:
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -83,153 +103,175 @@ void after() {
@Test
void testClosingWithoutInput() throws Exception {
try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, false, false, new LinkedHashMap<>(), new AtomicReference<>())) {
+ createTestHarness(
+ false, false, false, new LinkedHashMap<>(), new AtomicReference<>())) {
testHarness.setup();
testHarness.open();
}
}
@Test
void testNonPartition() throws Exception {
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, false, false, new LinkedHashMap<>(), ref)) {
- writeUnorderedRecords(testHarness);
- assertThat(getFileContentByPath(tmpPath)).hasSize(1);
- }
-
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- assertThat(content.values())
- .containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
- }
-
- private void writeUnorderedRecords(OneInputStreamOperatorTestHarness<Row, Object> testHarness)
- throws Exception {
- testHarness.setup();
- testHarness.open();
-
- testHarness.processElement(new StreamRecord<>(Row.of("a1", 1, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p2"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a3", 3, "p1"), 1L));
+ checkWriteAndCommit(
+ false,
+ false,
+ false,
+ new LinkedHashMap<>(),
+ DEFAULT_INPUT_SUPPLIER,
+ DEFAULT_OUTPUT_SUPPLIER);
}
@Test
void testOverrideNonPartition() throws Exception {
testNonPartition();
-
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(true, false, false, new LinkedHashMap<>(), ref)) {
- writeUnorderedRecords(testHarness);
- assertThat(getFileContentByPath(tmpPath)).hasSize(1);
- }
-
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- assertThat(content).hasSize(1);
- assertThat(content.values())
- .containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
- assertThat(new File(tmpPath.toUri())).doesNotExist();
+ checkWriteAndCommit(
+ true,
+ false,
+ false,
+ new LinkedHashMap<>(),
+ DEFAULT_INPUT_SUPPLIER,
+ DEFAULT_OUTPUT_SUPPLIER);
}
@Test
void testStaticPartition() throws Exception {
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
LinkedHashMap<String, String> staticParts = new LinkedHashMap<>();
staticParts.put("c", "p1");
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, true, false, staticParts, ref)) {
- testHarness.setup();
- testHarness.open();
-
- testHarness.processElement(new StreamRecord<>(Row.of("a1", 1), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a3", 3), 1L));
- assertThat(getFileContentByPath(tmpPath)).hasSize(1);
- }
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- assertThat(content).hasSize(1);
- assertThat(content.keySet().iterator().next().getParentFile().getName()).isEqualTo("c=p1");
- assertThat(content.values()).containsExactly("a1,1\n" + "a2,2\n" + "a2,2\n" + "a3,3\n");
- assertThat(new File(tmpPath.toUri())).doesNotExist();
+ checkWriteAndCommit(
+ false,
+ true,
+ false,
+ staticParts,
+ () ->
+ Arrays.asList(
+ new StreamRecord<>(Row.of("a1", 1), 1L),
+ new StreamRecord<>(Row.of("a2", 2), 1L),
+ new StreamRecord<>(Row.of("a2", 2), 1L),
+ new StreamRecord<>(Row.of("a3", 3), 1L)),
+ () ->
+ Collections.singletonList(
+ "c=p1:" + "a1,1\n" + "a2,2\n" + "a2,2\n" + "a3,3\n"));
}
@Test
void testDynamicPartition() throws Exception {
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, true, false, new LinkedHashMap<>(), ref)) {
- writeUnorderedRecords(testHarness);
- assertThat(getFileContentByPath(tmpPath)).hasSize(2);
- }
-
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- Map<String, String> sortedContent = new TreeMap<>();
- content.forEach((file, s) -> sortedContent.put(file.getParentFile().getName(), s));
-
- assertThat(sortedContent).hasSize(2);
- assertThat(sortedContent)
- .contains(entry("c=p1", "a1,1\n" + "a2,2\n" + "a3,3\n"), entry("c=p2", "a2,2\n"));
- assertThat(new File(tmpPath.toUri())).doesNotExist();
+ checkWriteAndCommit(
+ false,
+ true,
+ false,
+ new LinkedHashMap<>(),
+ DEFAULT_INPUT_SUPPLIER,
+ () -> Arrays.asList("c=p1:" + "a1,1\n" + "a2,2\n" + "a3,3\n", "c=p2:" + "a2,2\n"));
}
@Test
void testGroupedDynamicPartition() throws Exception {
+ checkWriteAndCommit(
+ false,
+ true,
+ true,
+ new LinkedHashMap<>(),
+ () ->
+ Arrays.asList(
+ new StreamRecord<>(Row.of("a1", 1, "p1"), 1L),
+ new StreamRecord<>(Row.of("a2", 2, "p1"), 1L),
+ new StreamRecord<>(Row.of("a3", 3, "p1"), 1L),
+ new StreamRecord<>(Row.of("a2", 2, "p2"), 1L)),
+ () -> Arrays.asList("c=p1:" + "a1,1\n" + "a2,2\n" + "a3,3\n", "c=p2:" + "a2,2\n"));
+ }
+
+ @Test
+ void testGetUniqueStagingDirectory() {
+ final int expectedStagingDirectories = 100;
+ for (int i = 0; i < expectedStagingDirectories; i++) {
+ assertDoesNotThrow(() -> createSinkFormat(false, false, false, new LinkedHashMap<>()));
+ }
+ assertThat(outputPath.toFile().listFiles()).hasSize(expectedStagingDirectories);
+ }
+
+ private void checkWriteAndCommit(
+ boolean override,
+ boolean partitioned,
+ boolean dynamicGrouped,
+ LinkedHashMap<String, String> staticPartitions,
+ Supplier<List<StreamRecord<Row>>> inputSupplier,
+ Supplier<List<String>> outputSupplier)
+ throws Exception {
+ List<String> expectedOutput = outputSupplier.get();
+ int expectedFileNum = expectedOutput.size();
AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, true, true, new LinkedHashMap<>(), ref)) {
+ createTestHarness(override, partitioned, dynamicGrouped, staticPartitions, ref)) {
testHarness.setup();
testHarness.open();
-
- testHarness.processElement(new StreamRecord<>(Row.of("a1", 1, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a3", 3, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p2"), 1L));
- assertThat(getFileContentByPath(tmpPath)).hasSize(2);
+ for (StreamRecord<Row> record : inputSupplier.get()) {
+ testHarness.processElement(record);
+ }
+ assertThat(getStagingFileContent(ref)).hasSize(expectedFileNum);
}
ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- Map<String, String> sortedContent = new TreeMap<>();
- content.forEach((file, s) -> sortedContent.put(file.getParentFile().getName(), s));
+ assertThat(Paths.get(ref.get().getStagingPath().getPath())).doesNotExist();
+
+ Map<File, String> fileToContent = getFileContentByPath(outputPath);
+ assertThat(fileToContent).hasSize(expectedFileNum);
+ if (partitioned) {
+ assertPartitionedOutput(expectedOutput, fileToContent);
+ } else {
+ assertThat(fileToContent.values()).containsExactlyInAnyOrderElementsOf(expectedOutput);
+ }
+ }
- assertThat(sortedContent).hasSize(2);
- assertThat(sortedContent.get("c=p1")).isEqualTo("a1,1\n" + "a2,2\n" + "a3,3\n");
- assertThat(sortedContent.get("c=p2")).isEqualTo("a2,2\n");
- assertThat(new File(tmpPath.toUri())).doesNotExist();
+ private void assertPartitionedOutput(
+ List<String> expectedOutput, Map<File, String> fileToContent) {
Review Comment:
```suggestion
Map<String, String> expectedOutput, Map<File, String> fileToContent) {
```
The expected output being encoded with a colon could be refactored by using a `Map<String, String>` as far as I can see. That would improve the readability of the test code. The sole purpose of the colon is to have the partition info and the content being stored together as far as I can see.
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -83,153 +103,175 @@ void after() {
@Test
void testClosingWithoutInput() throws Exception {
try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, false, false, new LinkedHashMap<>(), new AtomicReference<>())) {
+ createTestHarness(
+ false, false, false, new LinkedHashMap<>(), new AtomicReference<>())) {
testHarness.setup();
testHarness.open();
}
}
@Test
void testNonPartition() throws Exception {
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, false, false, new LinkedHashMap<>(), ref)) {
- writeUnorderedRecords(testHarness);
- assertThat(getFileContentByPath(tmpPath)).hasSize(1);
- }
-
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- assertThat(content.values())
- .containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
- }
-
- private void writeUnorderedRecords(OneInputStreamOperatorTestHarness<Row, Object> testHarness)
- throws Exception {
- testHarness.setup();
- testHarness.open();
-
- testHarness.processElement(new StreamRecord<>(Row.of("a1", 1, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p2"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a3", 3, "p1"), 1L));
+ checkWriteAndCommit(
+ false,
+ false,
+ false,
+ new LinkedHashMap<>(),
+ DEFAULT_INPUT_SUPPLIER,
+ DEFAULT_OUTPUT_SUPPLIER);
}
@Test
void testOverrideNonPartition() throws Exception {
testNonPartition();
-
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(true, false, false, new LinkedHashMap<>(), ref)) {
- writeUnorderedRecords(testHarness);
- assertThat(getFileContentByPath(tmpPath)).hasSize(1);
- }
-
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- assertThat(content).hasSize(1);
- assertThat(content.values())
- .containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
- assertThat(new File(tmpPath.toUri())).doesNotExist();
+ checkWriteAndCommit(
+ true,
+ false,
+ false,
+ new LinkedHashMap<>(),
+ DEFAULT_INPUT_SUPPLIER,
+ DEFAULT_OUTPUT_SUPPLIER);
}
@Test
void testStaticPartition() throws Exception {
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
LinkedHashMap<String, String> staticParts = new LinkedHashMap<>();
staticParts.put("c", "p1");
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, true, false, staticParts, ref)) {
- testHarness.setup();
- testHarness.open();
-
- testHarness.processElement(new StreamRecord<>(Row.of("a1", 1), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a3", 3), 1L));
- assertThat(getFileContentByPath(tmpPath)).hasSize(1);
- }
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- assertThat(content).hasSize(1);
- assertThat(content.keySet().iterator().next().getParentFile().getName()).isEqualTo("c=p1");
- assertThat(content.values()).containsExactly("a1,1\n" + "a2,2\n" + "a2,2\n" + "a3,3\n");
- assertThat(new File(tmpPath.toUri())).doesNotExist();
+ checkWriteAndCommit(
+ false,
+ true,
+ false,
+ staticParts,
+ () ->
+ Arrays.asList(
+ new StreamRecord<>(Row.of("a1", 1), 1L),
+ new StreamRecord<>(Row.of("a2", 2), 1L),
+ new StreamRecord<>(Row.of("a2", 2), 1L),
+ new StreamRecord<>(Row.of("a3", 3), 1L)),
+ () ->
+ Collections.singletonList(
+ "c=p1:" + "a1,1\n" + "a2,2\n" + "a2,2\n" + "a3,3\n"));
}
@Test
void testDynamicPartition() throws Exception {
- AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
- try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, true, false, new LinkedHashMap<>(), ref)) {
- writeUnorderedRecords(testHarness);
- assertThat(getFileContentByPath(tmpPath)).hasSize(2);
- }
-
- ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- Map<String, String> sortedContent = new TreeMap<>();
- content.forEach((file, s) -> sortedContent.put(file.getParentFile().getName(), s));
-
- assertThat(sortedContent).hasSize(2);
- assertThat(sortedContent)
- .contains(entry("c=p1", "a1,1\n" + "a2,2\n" + "a3,3\n"), entry("c=p2", "a2,2\n"));
- assertThat(new File(tmpPath.toUri())).doesNotExist();
+ checkWriteAndCommit(
+ false,
+ true,
+ false,
+ new LinkedHashMap<>(),
+ DEFAULT_INPUT_SUPPLIER,
+ () -> Arrays.asList("c=p1:" + "a1,1\n" + "a2,2\n" + "a3,3\n", "c=p2:" + "a2,2\n"));
}
@Test
void testGroupedDynamicPartition() throws Exception {
+ checkWriteAndCommit(
+ false,
+ true,
+ true,
+ new LinkedHashMap<>(),
+ () ->
+ Arrays.asList(
+ new StreamRecord<>(Row.of("a1", 1, "p1"), 1L),
+ new StreamRecord<>(Row.of("a2", 2, "p1"), 1L),
+ new StreamRecord<>(Row.of("a3", 3, "p1"), 1L),
+ new StreamRecord<>(Row.of("a2", 2, "p2"), 1L)),
+ () -> Arrays.asList("c=p1:" + "a1,1\n" + "a2,2\n" + "a3,3\n", "c=p2:" + "a2,2\n"));
+ }
+
+ @Test
+ void testGetUniqueStagingDirectory() {
+ final int expectedStagingDirectories = 100;
+ for (int i = 0; i < expectedStagingDirectories; i++) {
+ assertDoesNotThrow(() -> createSinkFormat(false, false, false, new LinkedHashMap<>()));
+ }
+ assertThat(outputPath.toFile().listFiles()).hasSize(expectedStagingDirectories);
+ }
+
+ private void checkWriteAndCommit(
+ boolean override,
+ boolean partitioned,
+ boolean dynamicGrouped,
+ LinkedHashMap<String, String> staticPartitions,
+ Supplier<List<StreamRecord<Row>>> inputSupplier,
+ Supplier<List<String>> outputSupplier)
+ throws Exception {
+ List<String> expectedOutput = outputSupplier.get();
+ int expectedFileNum = expectedOutput.size();
AtomicReference<FileSystemOutputFormat<Row>> ref = new AtomicReference<>();
try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
- createSink(false, true, true, new LinkedHashMap<>(), ref)) {
+ createTestHarness(override, partitioned, dynamicGrouped, staticPartitions, ref)) {
testHarness.setup();
testHarness.open();
-
- testHarness.processElement(new StreamRecord<>(Row.of("a1", 1, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a3", 3, "p1"), 1L));
- testHarness.processElement(new StreamRecord<>(Row.of("a2", 2, "p2"), 1L));
- assertThat(getFileContentByPath(tmpPath)).hasSize(2);
+ for (StreamRecord<Row> record : inputSupplier.get()) {
+ testHarness.processElement(record);
+ }
+ assertThat(getStagingFileContent(ref)).hasSize(expectedFileNum);
}
ref.get().finalizeGlobal(finalizationContext);
- Map<File, String> content = getFileContentByPath(outputPath);
- Map<String, String> sortedContent = new TreeMap<>();
- content.forEach((file, s) -> sortedContent.put(file.getParentFile().getName(), s));
+ assertThat(Paths.get(ref.get().getStagingPath().getPath())).doesNotExist();
+
+ Map<File, String> fileToContent = getFileContentByPath(outputPath);
+ assertThat(fileToContent).hasSize(expectedFileNum);
+ if (partitioned) {
+ assertPartitionedOutput(expectedOutput, fileToContent);
+ } else {
+ assertThat(fileToContent.values()).containsExactlyInAnyOrderElementsOf(expectedOutput);
+ }
+ }
- assertThat(sortedContent).hasSize(2);
- assertThat(sortedContent.get("c=p1")).isEqualTo("a1,1\n" + "a2,2\n" + "a3,3\n");
- assertThat(sortedContent.get("c=p2")).isEqualTo("a2,2\n");
- assertThat(new File(tmpPath.toUri())).doesNotExist();
+ private void assertPartitionedOutput(
+ List<String> expectedOutput, Map<File, String> fileToContent) {
+ final String delimiter = ":";
+ Map<String, String> partitionToContent =
+ fileToContent.entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ e -> e.getKey().getParentFile().getName(),
+ Map.Entry::getValue));
+
+ Map<String, String> expectedPartitionContent =
+ expectedOutput.stream()
+ .map(r -> r.split(delimiter))
+ .collect(Collectors.toMap(splits -> splits[0], splits -> splits[1]));
+
+ assertThat(partitionToContent).containsExactlyInAnyOrderEntriesOf(expectedPartitionContent);
}
- private OneInputStreamOperatorTestHarness<Row, Object> createSink(
+ private FileSystemOutputFormat<Row> createSinkFormat(
boolean override,
boolean partition,
boolean dynamicGrouped,
- LinkedHashMap<String, String> staticPartitions,
- AtomicReference<FileSystemOutputFormat<Row>> sinkRef)
- throws Exception {
+ LinkedHashMap<String, String> staticPartitions) {
String[] columnNames = new String[] {"a", "b", "c"};
String[] partitionColumns = partition ? new String[] {"c"} : new String[0];
+ Path path = new Path(outputPath.toString());
+ TableMetaStoreFactory msFactory = new FileSystemCommitterTest.TestMetaStoreFactory(path);
+ return new FileSystemOutputFormat.Builder<Row>()
+ .setMetaStoreFactory(msFactory)
+ .setPath(path)
+ .setOverwrite(override)
+ .setPartitionColumns(partitionColumns)
+ .setPartitionComputer(
+ new RowPartitionComputer("default", columnNames, partitionColumns))
+ .setFormatFactory(TextOutputFormat::new)
+ .setDynamicGrouped(dynamicGrouped)
+ .setStaticPartitions(staticPartitions)
+ .build();
+ }
- TableMetaStoreFactory msFactory =
- new FileSystemCommitterTest.TestMetaStoreFactory(new Path(outputPath.toString()));
+ private OneInputStreamOperatorTestHarness<Row, Object> createTestHarness(
+ boolean override,
+ boolean partition,
+ boolean dynamicGrouped,
+ LinkedHashMap<String, String> staticPartitions,
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef)
+ throws Exception {
Review Comment:
```suggestion
private OneInputStreamOperatorTestHarness<Row, Object> createTestHarness(
OutputFormat<Row> outputFormat)
throws Exception {
```
The `AtomicReference` is actually not needed here. You're forwarding all the parameters of this method to `createSinkFormat` and pass the `AtomicReference` to capture the created instance. You could just create the `FileSystemOutputFormat` instance outside of this method and pass it into it achieving the same outcome.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][table-planner][filesystem][test] Fix unstable TableSourceITCase#testTableHintWithLogicalTableScanReuse [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on PR #24390:
URL: https://github.com/apache/flink/pull/24390#issuecomment-1966220956
> Thanks for looking into, @LadyForest. That also means that it's not a test issue but a production code issue, isn't it? The commit message is misleading in that case. Is there a way to add a junit test to cover this scenario?
>
> I verified that the test instability is fixed with this issue 👍
Sure. Let me add a UT case to verify this fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on PR #24390:
URL: https://github.com/apache/flink/pull/24390#issuecomment-1970441742
Hi @snuyanzin @XComp, thanks for your comments on the review. I updated the test (actually not perfect either..) and tried best to verify the case. Would you mind retaking a look?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1520768668
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemOutputFormat.java:
##########
@@ -286,10 +302,21 @@ public Builder<T> setIsToLocal(boolean isToLocal) {
}
public Builder<T> setPath(Path path) {
+ return this.setStagingPath(toStagingPath(path));
+ }
Review Comment:
Nit: to avoid arch rule being violated
```suggestion
this.path = toStagingPath(path);
return this;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on PR #24390:
URL: https://github.com/apache/flink/pull/24390#issuecomment-1993628470
Hi @XComp, I plan to merge PR since the CI passed unless you have any other concerns.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on PR #24390:
URL: https://github.com/apache/flink/pull/24390#issuecomment-1993735441
No, go ahead. It would be good to provide backports as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1520688625
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemOutputFormat.java:
##########
@@ -197,13 +200,37 @@ public void close() throws IOException {
}
}
+ private Path toStagingPath(Path path) {
+ // Add a random UUID to prevent multiple sinks from sharing the same staging dir.
+ // Please see FLINK-29114 for more details
Review Comment:
+1. Let me remove this comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1507274806
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -104,15 +113,101 @@ void testNonPartition() throws Exception {
.containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testStagingDirBehavior(boolean shareStagingDir) throws Exception {
+ // sink1
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef1 = new AtomicReference<>();
+ AtomicReference<Map<File, String>> fileToCommitRef1 = new AtomicReference<>();
+ writeRowsToSink(
+ sinkRef1,
+ fileToCommitRef1,
+ getStagingDir(shareStagingDir),
+ Row.of("a1", 1, "x1"),
+ Row.of("a2", 2, "x2"));
+
+ // sink2
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef2 = new AtomicReference<>();
+ AtomicReference<Map<File, String>> fileToCommitRef2 = new AtomicReference<>();
+ writeRowsToSink(
+ sinkRef2,
+ fileToCommitRef2,
+ getStagingDir(shareStagingDir),
+ Row.of("b1", 1, "y1"),
+ Row.of("b2", 2, "y2"));
+
+ assertSinkBehavior(sinkRef1, fileToCommitRef1, sinkRef2, fileToCommitRef2, shareStagingDir);
+ }
+
+ private void writeRowsToSink(
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef,
+ AtomicReference<Map<File, String>> contentRef,
+ Path stagingDir,
+ Row... records)
+ throws Exception {
+ try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
+ createSink(false, false, false, stagingDir, new LinkedHashMap<>(), sinkRef)) {
+ writeUnorderedRecords(testHarness, Arrays.asList(records));
+ contentRef.set(getFileContentByPath(Paths.get(stagingDir.getPath())));
+ }
+ }
+
+ private Path getStagingDir(boolean shareStagingDir) {
+ String pathPostfix = FileSystemTableSink.getStagingPathPostfix(shareStagingDir);
+ return Path.fromLocalFile(tmpPath.resolve(pathPostfix).toFile());
+ }
+
+ private void assertSinkBehavior(
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef1,
+ AtomicReference<Map<File, String>> fileToCommitRef1,
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef2,
+ AtomicReference<Map<File, String>> fileToCommitRef2,
+ boolean shareStagingDir)
+ throws Exception {
+ Map<File, String> fileToCommit1 = fileToCommitRef1.get();
+ Map<File, String> fileToCommit2 = fileToCommitRef2.get();
+ assertThat(fileToCommit2.keySet()).allMatch(File::exists);
+ if (shareStagingDir) {
+ assertThat(fileToCommit1.keySet()).noneMatch(File::exists);
+ } else {
+ assertThat(fileToCommit1.keySet()).allMatch(File::exists);
+ }
+ sinkRef1.get().finalizeGlobal(finalizationContext);
+ sinkRef2.get().finalizeGlobal(finalizationContext);
+ Map<File, String> committedFiles = getFileContentByPath(outputPath);
+ if (shareStagingDir) {
Review Comment:
Your understanding is correct. It's just that this test is actually testing `FileSystemOutputFormat` for a scenario where some other class (in our case `FileSystemTableSink`) has a bug.
This test is fine if you think that `FileSystemOutputFormat` should be able to handle this case. But I'm wondering whether it would be more appropriate to fail in such a case where multiple instances are accessing the same folder (but there's the question how easily a `FileSystemOutputFormat` instance can detect something like that.
Anyway, resolving this test scenario easily indicates that the scope of the different classes is not well-defined.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1521155291
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemOutputFormat.java:
##########
@@ -96,6 +99,22 @@ private FileSystemOutputFormat(
this.outputFileConfig = outputFileConfig;
this.identifier = identifier;
this.partitionCommitPolicyFactory = partitionCommitPolicyFactory;
+
+ createStagingDirectory(this.stagingPath);
+ }
+
+ private static void createStagingDirectory(Path stagingPath) {
+ try {
+ final FileSystem stagingFileSystem = stagingPath.getFileSystem();
+ Preconditions.checkState(
+ !stagingFileSystem.exists(stagingPath),
+ "Staging dir %s already exists",
+ stagingFileSystem);
Review Comment:
Nit: just noticed a typo. It should be `stagingPath` instead of `stagingFileSystem` at L#112
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1508526295
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/utils/PathUtils.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.file.table.utils;
+
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.UUID;
+
+/** Path utils for file system. */
+public class PathUtils {
Review Comment:
I originally planned to move the logic for generating the staging directory into `FileSystemOutputFormat`. However, when modifying the `HiveTableSink`, I found that `createBatchCompactSink` relies on tmpPath.
I didn't want to introduce changes to other modules, so I ultimately opted for a compromise and extracted this utility method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][table-planner][filesystem][test] Fix unstable TableSourceITCase#testTableHintWithLogicalTableScanReuse [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1503917055
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java:
##########
@@ -374,7 +374,8 @@ public DynamicTableSource.DataStructureConverter createDataStructureConverter(
}
private Path toStagingPath() {
- Path stagingDir = new Path(path, ".staging_" + System.currentTimeMillis());
+ Path stagingDir =
+ new Path(path, ".staging_" + UUID.randomUUID() + System.currentTimeMillis());
Review Comment:
I guess, we don't need the `System.currentMillis()` if we use UUIDs, do we?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on PR #24390:
URL: https://github.com/apache/flink/pull/24390#issuecomment-1976845985
Hi @XComp, would you mind taking a look when you're available?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1505818193
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java:
##########
@@ -374,7 +374,10 @@ public DynamicTableSource.DataStructureConverter createDataStructureConverter(
}
private Path toStagingPath() {
- Path stagingDir = new Path(path, ".staging_" + System.currentTimeMillis());
+ // Add a random UUID to prevent multiple sinks from sharing the same staging dir.
+ // Please see FLINK-29114 for more details
+ Path stagingDir =
+ new Path(path, ".staging_" + UUID.randomUUID() + System.currentTimeMillis());
Review Comment:
The last two sentences are contradicting. ;-) But I'm fine if you think that the timestamp is helpful. But you might want to add a separator between the time and the UUID. And you might want to put the timestamp before the UUID if you think it's helpful for troubleshooting (for the sake of ordering by file name).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-29114][connector][filesystem] Fix issue of file overwriting caused by multiple writes to the same sink table and shared staging directory [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #24390:
URL: https://github.com/apache/flink/pull/24390#discussion_r1507187446
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemOutputFormatTest.java:
##########
@@ -104,15 +113,101 @@ void testNonPartition() throws Exception {
.containsExactly("a1,1,p1\n" + "a2,2,p1\n" + "a2,2,p2\n" + "a3,3,p1\n");
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testStagingDirBehavior(boolean shareStagingDir) throws Exception {
+ // sink1
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef1 = new AtomicReference<>();
+ AtomicReference<Map<File, String>> fileToCommitRef1 = new AtomicReference<>();
+ writeRowsToSink(
+ sinkRef1,
+ fileToCommitRef1,
+ getStagingDir(shareStagingDir),
+ Row.of("a1", 1, "x1"),
+ Row.of("a2", 2, "x2"));
+
+ // sink2
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef2 = new AtomicReference<>();
+ AtomicReference<Map<File, String>> fileToCommitRef2 = new AtomicReference<>();
+ writeRowsToSink(
+ sinkRef2,
+ fileToCommitRef2,
+ getStagingDir(shareStagingDir),
+ Row.of("b1", 1, "y1"),
+ Row.of("b2", 2, "y2"));
+
+ assertSinkBehavior(sinkRef1, fileToCommitRef1, sinkRef2, fileToCommitRef2, shareStagingDir);
+ }
+
+ private void writeRowsToSink(
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef,
+ AtomicReference<Map<File, String>> contentRef,
+ Path stagingDir,
+ Row... records)
+ throws Exception {
+ try (OneInputStreamOperatorTestHarness<Row, Object> testHarness =
+ createSink(false, false, false, stagingDir, new LinkedHashMap<>(), sinkRef)) {
+ writeUnorderedRecords(testHarness, Arrays.asList(records));
+ contentRef.set(getFileContentByPath(Paths.get(stagingDir.getPath())));
+ }
+ }
+
+ private Path getStagingDir(boolean shareStagingDir) {
+ String pathPostfix = FileSystemTableSink.getStagingPathPostfix(shareStagingDir);
+ return Path.fromLocalFile(tmpPath.resolve(pathPostfix).toFile());
+ }
+
+ private void assertSinkBehavior(
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef1,
+ AtomicReference<Map<File, String>> fileToCommitRef1,
+ AtomicReference<FileSystemOutputFormat<Row>> sinkRef2,
+ AtomicReference<Map<File, String>> fileToCommitRef2,
+ boolean shareStagingDir)
+ throws Exception {
+ Map<File, String> fileToCommit1 = fileToCommitRef1.get();
+ Map<File, String> fileToCommit2 = fileToCommitRef2.get();
+ assertThat(fileToCommit2.keySet()).allMatch(File::exists);
+ if (shareStagingDir) {
+ assertThat(fileToCommit1.keySet()).noneMatch(File::exists);
+ } else {
+ assertThat(fileToCommit1.keySet()).allMatch(File::exists);
+ }
+ sinkRef1.get().finalizeGlobal(finalizationContext);
+ sinkRef2.get().finalizeGlobal(finalizationContext);
+ Map<File, String> committedFiles = getFileContentByPath(outputPath);
+ if (shareStagingDir) {
Review Comment:
I understand it this way, and please correct me if I'm wrong: When verifying the correctness of a fix, it's common to build some test cases to reliably reproduce the issue, then propose a solution to address that problem, and rerun the tests to ensure that the fix is appropriate and has not introduced any new issues. Coming back to this case, since `System.currentTimeMillis()` is a non-deterministic function in the original method, in order to reproduce the issue stably, it is necessary to make it return a constant value, which is why an additional boolean variable was introduced. However, after noticing your comments in another pull request, I think your approach is more reasonable, as it would throw an exception at compile time rather than delaying the problem to runtime, which could cause unexpected consequences.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org