[FLINK-29305][tests] UpsertTestSinkWriter creates parent directories

huanhaihu · Sep 15, 2022 · 33afc3c · 33afc3c
1 parent d3513d9
commit 33afc3c
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 1 deletion.
diff --git a/...-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSinkWriter.java b/...-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSinkWriter.java
@@ -27,6 +27,7 @@
 import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.nio.file.Files;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -53,6 +54,11 @@ class UpsertTestSinkWriter<IN> implements SinkWriter<IN> {
         this.keySerializationSchema = checkNotNull(keySerializationSchema);
         this.valueSerializationSchema = checkNotNull(valueSerializationSchema);
         checkNotNull(outputFile);
+        try {
+            Files.createDirectories(outputFile.toPath().getParent());
+        } catch (IOException e) {
+            throw new FlinkRuntimeException("Could not parent directories for path: " + outputFile);
+        }
         try {
             this.bufferedOutputStream =
                     new BufferedOutputStream(new FileOutputStream(outputFile, true));

diff --git a/.../src/test/java/org/apache/flink/connector/upserttest/sink/UpsertTestSinkWriterITCase.java b/.../src/test/java/org/apache/flink/connector/upserttest/sink/UpsertTestSinkWriterITCase.java
@@ -35,6 +35,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -60,7 +61,7 @@ class UpsertTestSinkWriterITCase {
 
     @BeforeEach
     void setup() {
-        outputFile = new File(tempDir, "records.out");
+        outputFile = tempDir.toPath().resolve(Paths.get("dir", "records.out")).toFile();
         writer = createSinkWriter(outputFile);
         expectedRecords = writeTestData(writer);
     }