[FLINK-28513] Fix Flink Table API CSV streaming sink throws Serialize…

…dThrowable exception
nicusX · Sep 4, 2023 · e921489 · e921489
1 parent fbef3c2
commit e921489
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 1 deletion.
diff --git a/...e/src/main/java/org/apache/flink/fs/s3/common/writer/S3RecoverableFsDataOutputStream.java b/...e/src/main/java/org/apache/flink/fs/s3/common/writer/S3RecoverableFsDataOutputStream.java
@@ -126,7 +126,16 @@ public long getPos() throws IOException {
 
     @Override
     public void sync() throws IOException {
-        fileStream.sync();
+        lock();
+        try {
+            fileStream.flush();
+            openNewPartIfNecessary(userDefinedMinPartSize);
+            Committer committer = upload.snapshotAndGetCommitter();
+            committer.commitAfterRecovery();
+            closeForCommit();
+        } finally {
+            unlock();
+        }
     }
 
     @Override

diff --git a/...c/test/java/org/apache/flink/fs/s3/common/writer/S3RecoverableFsDataOutputStreamTest.java b/...c/test/java/org/apache/flink/fs/s3/common/writer/S3RecoverableFsDataOutputStreamTest.java
@@ -254,6 +254,16 @@ public void closeForCommitOnClosedStreamShouldFail() throws IOException {
         streamUnderTest.closeForCommit().commit();
     }
 
+    @Test(expected = Exception.class)
+    public void testSync() throws IOException {
+        streamUnderTest.write(bytesOf("hello"));
+        streamUnderTest.write(bytesOf(" world"));
+        streamUnderTest.sync();
+        assertThat(multipartUploadUnderTest, hasContent(bytesOf("hello world")));
+        streamUnderTest.write(randomBuffer(RefCountedBufferingFileStream.BUFFER_SIZE + 1));
+        assertThat(multipartUploadUnderTest, hasContent(bytesOf("hello world")));
+    }
+
     // ------------------------------------------------------------------------------------------------------------
     // Utils
     // ------------------------------------------------------------------------------------------------------------