This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 5b2d8a491a [Fix][Connector-V2] Fix file sink checkpoint write race
(#11103)
5b2d8a491a is described below
commit 5b2d8a491aa9ec5514016ee56ca9ef4e8e31385d
Author: Jast <[email protected]>
AuthorDate: Sun Oct 11 09:46:27 2026 +0000
[Fix][Connector-V2] Fix file sink checkpoint write race (#11103)
Co-authored-by: davidzollo <[email protected]>
---
.../file/sink/writer/AbstractWriteStrategy.java | 24 ++-
.../file/sink/writer/BinaryWriteStrategy.java | 9 +-
.../file/sink/writer/CanalJsonWriteStrategy.java | 4 +-
.../file/sink/writer/CsvWriteStrategy.java | 4 +-
.../sink/writer/DebeziumJsonWriteStrategy.java | 4 +-
.../file/sink/writer/ExcelWriteStrategy.java | 4 +-
.../file/sink/writer/JsonWriteStrategy.java | 4 +-
.../file/sink/writer/MaxWellJsonWriteStrategy.java | 4 +-
.../file/sink/writer/OrcWriteStrategy.java | 4 +-
.../file/sink/writer/ParquetWriteStrategy.java | 4 +-
.../file/sink/writer/TextWriteStrategy.java | 4 +-
.../file/sink/writer/XmlWriteStrategy.java | 4 +-
.../sink/writer/AbstractWriteStrategyTest.java | 219 +++++++++++++++++++++
13 files changed, 263 insertions(+), 29 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/AbstractWriteStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/AbstractWriteStrategy.java
index edd1fe6c0b..591391a82d 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/AbstractWriteStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/AbstractWriteStrategy.java
@@ -72,6 +72,15 @@ import java.util.UUID;
import java.util.regex.Matcher;
import java.util.stream.Collectors;
+/**
+ * Concurrency contract: {@code write}/{@code finishAndCloseFile} (declared
per-format in
+ * subclasses) and {@code prepareCommit}/{@code snapshotState}/{@code
applySchemaChange}/{@code
+ * abortPrepare} (here in the base class) all synchronize on the strategy
instance monitor so that a
+ * checkpoint can never close, snapshot, or delete a transaction's files while
a row write is still
+ * in flight (see #10523). Java does not require an override to keep a
superclass method's {@code
+ * synchronized} modifier, so every subclass that overrides one of these
methods MUST also declare
+ * it {@code synchronized}, or it silently drops out of this lock and reopens
the race.
+ */
public abstract class AbstractWriteStrategy<T> implements WriteStrategy<T> {
protected final Logger log = LoggerFactory.getLogger(this.getClass());
protected final FileSinkConfig fileSinkConfig;
@@ -227,7 +236,7 @@ public abstract class AbstractWriteStrategy<T> implements
WriteStrategy<T> {
// ── Schema Evolution
──────────────────────────────────────────────────────────
@Override
- public void applySchemaChange(SchemaChangeEvent event) throws IOException {
+ public synchronized void applySchemaChange(SchemaChangeEvent event) throws
IOException {
if (!fileSinkConfig.isSchemaEvolutionEnabled()
&& !(event instanceof RestoreTableSchemaEvent)) {
throw new UnsupportedOperationException(
@@ -580,7 +589,7 @@ public abstract class AbstractWriteStrategy<T> implements
WriteStrategy<T> {
*/
@SneakyThrows
@Override
- public Optional<FileCommitInfo> prepareCommit() {
+ public synchronized Optional<FileCommitInfo> prepareCommit() {
if (this.needMoveFiles.isEmpty() &&
fileSinkConfig.isCreateEmptyFileWhenNoData()) {
String filePath = createFilePathWithoutPartition();
this.getOrCreateOutputStream(filePath);
@@ -600,16 +609,19 @@ public abstract class AbstractWriteStrategy<T> implements
WriteStrategy<T> {
/** abort prepare commit operation */
@Override
- public void abortPrepare() {
+ public synchronized void abortPrepare() {
abortPrepare(transactionId);
}
/**
- * abort prepare commit operation using transaction directory
+ * Abort prepare commit operation using transaction directory.
Synchronized so a checkpoint
+ * abort cannot delete the transaction directory (and clear in-flight file
bookkeeping) while a
+ * row write is still in progress on this strategy instance; see the
class-level concurrency
+ * contract note.
*
* @param transactionId transaction id
*/
- public void abortPrepare(String transactionId) {
+ public synchronized void abortPrepare(String transactionId) {
String transactionDir = getTransactionDir(transactionId);
try {
hadoopFileSystemProxy.deleteFile(transactionDir);
@@ -678,7 +690,7 @@ public abstract class AbstractWriteStrategy<T> implements
WriteStrategy<T> {
* @return the list of states
*/
@Override
- public List<FileSinkState> snapshotState(long checkpointId) {
+ public synchronized List<FileSinkState> snapshotState(long checkpointId) {
LinkedHashMap<String, List<String>> commitMap =
this.partitionDirAndValuesMap.entrySet().stream()
.collect(
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/BinaryWriteStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/BinaryWriteStrategy.java
index eb27a17a9a..b1212d5e74 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/BinaryWriteStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/BinaryWriteStrategy.java
@@ -63,7 +63,7 @@ public class BinaryWriteStrategy extends
AbstractWriteStrategy<FSDataOutputStrea
}
@Override
- public void write(SeaTunnelRow seaTunnelRow) throws FileConnectorException
{
+ public synchronized void write(SeaTunnelRow seaTunnelRow) throws
FileConnectorException {
long partIndex = (long) seaTunnelRow.getField(2);
if (partIndex == -1) {
return;
@@ -143,8 +143,11 @@ public class BinaryWriteStrategy extends
AbstractWriteStrategy<FSDataOutputStrea
return fsDataOutputStream;
}
+ // synchronized to stay on the strategy-instance lock declared by the
overridden
+ // AbstractWriteStrategy#applySchemaChange contract; `synchronized` is not
inherited by
+ // overrides in Java, so dropping it here would silently exit that lock.
@Override
- public void applySchemaChange(SchemaChangeEvent event) {
+ public synchronized void applySchemaChange(SchemaChangeEvent event) {
throw new FileConnectorException(
FileConnectorErrorCode.FORMAT_NOT_SUPPORT,
"BinaryWriteStrategy does not support schema evolution. "
@@ -152,7 +155,7 @@ public class BinaryWriteStrategy extends
AbstractWriteStrategy<FSDataOutputStrea
}
@Override
- public void finishAndCloseFile() {
+ public synchronized void finishAndCloseFile() {
beingWrittenOutputStream.forEach(
(key, value) -> {
try {
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/CanalJsonWriteStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/CanalJsonWriteStrategy.java
index a7e1082d15..4c4916c10e 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/CanalJsonWriteStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/CanalJsonWriteStrategy.java
@@ -77,7 +77,7 @@ public class CanalJsonWriteStrategy extends
AbstractWriteStrategy<FSDataOutputSt
}
@Override
- public void write(@NonNull SeaTunnelRow seaTunnelRow) {
+ public synchronized void write(@NonNull SeaTunnelRow seaTunnelRow) {
super.write(seaTunnelRow);
String filePath = getOrCreateFilePathBeingWritten(seaTunnelRow);
FSDataOutputStream fsDataOutputStream =
getOrCreateOutputStream(filePath);
@@ -98,7 +98,7 @@ public class CanalJsonWriteStrategy extends
AbstractWriteStrategy<FSDataOutputSt
}
@Override
- public void finishAndCloseFile() {
+ public synchronized void finishAndCloseFile() {
beingWrittenOutputStream.forEach(
(key, value) -> {
try {
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/CsvWriteStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/CsvWriteStrategy.java
index 74e9bd0789..ca4513f900 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/CsvWriteStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/CsvWriteStrategy.java
@@ -107,7 +107,7 @@ public class CsvWriteStrategy extends
AbstractWriteStrategy<FSDataOutputStream>
}
@Override
- public void write(@NonNull SeaTunnelRow seaTunnelRow) {
+ public synchronized void write(@NonNull SeaTunnelRow seaTunnelRow) {
super.write(seaTunnelRow);
String filePath = getOrCreateFilePathBeingWritten(seaTunnelRow);
FSDataOutputStream fsDataOutputStream =
getOrCreateOutputStream(filePath);
@@ -124,7 +124,7 @@ public class CsvWriteStrategy extends
AbstractWriteStrategy<FSDataOutputStream>
}
@Override
- public void finishAndCloseFile() {
+ public synchronized void finishAndCloseFile() {
beingWrittenOutputStream.forEach(
(key, value) -> {
try {
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/DebeziumJsonWriteStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/DebeziumJsonWriteStrategy.java
index 8c26b57c30..0897dc913d 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/DebeziumJsonWriteStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/DebeziumJsonWriteStrategy.java
@@ -77,7 +77,7 @@ public class DebeziumJsonWriteStrategy extends
AbstractWriteStrategy<FSDataOutpu
}
@Override
- public void write(@NonNull SeaTunnelRow seaTunnelRow) {
+ public synchronized void write(@NonNull SeaTunnelRow seaTunnelRow) {
super.write(seaTunnelRow);
String filePath = getOrCreateFilePathBeingWritten(seaTunnelRow);
FSDataOutputStream fsDataOutputStream =
getOrCreateOutputStream(filePath);
@@ -98,7 +98,7 @@ public class DebeziumJsonWriteStrategy extends
AbstractWriteStrategy<FSDataOutpu
}
@Override
- public void finishAndCloseFile() {
+ public synchronized void finishAndCloseFile() {
beingWrittenOutputStream.forEach(
(key, value) -> {
try {
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ExcelWriteStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ExcelWriteStrategy.java
index 0fa1d260b4..0e61c5a7fd 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ExcelWriteStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ExcelWriteStrategy.java
@@ -38,7 +38,7 @@ public class ExcelWriteStrategy extends
AbstractWriteStrategy<ExcelGenerator> {
}
@Override
- public void write(SeaTunnelRow seaTunnelRow) {
+ public synchronized void write(SeaTunnelRow seaTunnelRow) {
super.write(seaTunnelRow);
String filePath = getOrCreateFilePathBeingWritten(seaTunnelRow);
ExcelGenerator excelGenerator = getOrCreateOutputStream(filePath);
@@ -46,7 +46,7 @@ public class ExcelWriteStrategy extends
AbstractWriteStrategy<ExcelGenerator> {
}
@Override
- public void finishAndCloseFile() {
+ public synchronized void finishAndCloseFile() {
this.beingWrittenWriter.forEach(
(k, v) -> {
try {
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/JsonWriteStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/JsonWriteStrategy.java
index 4b4bf421f2..ff858145a1 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/JsonWriteStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/JsonWriteStrategy.java
@@ -72,7 +72,7 @@ public class JsonWriteStrategy extends
AbstractWriteStrategy<FSDataOutputStream>
}
@Override
- public void write(@NonNull SeaTunnelRow seaTunnelRow) {
+ public synchronized void write(@NonNull SeaTunnelRow seaTunnelRow) {
super.write(seaTunnelRow);
String filePath = getOrCreateFilePathBeingWritten(seaTunnelRow);
FSDataOutputStream fsDataOutputStream =
getOrCreateOutputStream(filePath);
@@ -90,7 +90,7 @@ public class JsonWriteStrategy extends
AbstractWriteStrategy<FSDataOutputStream>
}
@Override
- public void finishAndCloseFile() {
+ public synchronized void finishAndCloseFile() {
beingWrittenOutputStream.forEach(
(key, value) -> {
try {
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/MaxWellJsonWriteStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/MaxWellJsonWriteStrategy.java
index bf69d02618..48e8e043f3 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/MaxWellJsonWriteStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/MaxWellJsonWriteStrategy.java
@@ -77,7 +77,7 @@ public class MaxWellJsonWriteStrategy extends
AbstractWriteStrategy<FSDataOutput
}
@Override
- public void write(@NonNull SeaTunnelRow seaTunnelRow) {
+ public synchronized void write(@NonNull SeaTunnelRow seaTunnelRow) {
super.write(seaTunnelRow);
String filePath = getOrCreateFilePathBeingWritten(seaTunnelRow);
FSDataOutputStream fsDataOutputStream =
getOrCreateOutputStream(filePath);
@@ -98,7 +98,7 @@ public class MaxWellJsonWriteStrategy extends
AbstractWriteStrategy<FSDataOutput
}
@Override
- public void finishAndCloseFile() {
+ public synchronized void finishAndCloseFile() {
beingWrittenOutputStream.forEach(
(key, value) -> {
try {
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/OrcWriteStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/OrcWriteStrategy.java
index 6dfc04d537..f2e61480f3 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/OrcWriteStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/OrcWriteStrategy.java
@@ -71,7 +71,7 @@ public class OrcWriteStrategy extends
AbstractWriteStrategy<Writer> {
}
@Override
- public void write(@NonNull SeaTunnelRow seaTunnelRow) {
+ public synchronized void write(@NonNull SeaTunnelRow seaTunnelRow) {
super.write(seaTunnelRow);
String filePath = getOrCreateFilePathBeingWritten(seaTunnelRow);
Writer writer = getOrCreateOutputStream(filePath);
@@ -103,7 +103,7 @@ public class OrcWriteStrategy extends
AbstractWriteStrategy<Writer> {
}
@Override
- public void finishAndCloseFile() {
+ public synchronized void finishAndCloseFile() {
List<FileConnectorException> closeErrors = new ArrayList<>();
this.beingWrittenWriter.forEach(
(k, v) -> {
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ParquetWriteStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ParquetWriteStrategy.java
index 53bd50692e..878c900e47 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ParquetWriteStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ParquetWriteStrategy.java
@@ -120,7 +120,7 @@ public class ParquetWriteStrategy extends
AbstractWriteStrategy<ParquetWriter<Ge
}
@Override
- public void write(@NonNull SeaTunnelRow seaTunnelRow) {
+ public synchronized void write(@NonNull SeaTunnelRow seaTunnelRow) {
super.write(seaTunnelRow);
String filePath = getOrCreateFilePathBeingWritten(seaTunnelRow);
ParquetWriter<GenericRecord> writer =
getOrCreateOutputStream(filePath);
@@ -148,7 +148,7 @@ public class ParquetWriteStrategy extends
AbstractWriteStrategy<ParquetWriter<Ge
}
@Override
- public void finishAndCloseFile() {
+ public synchronized void finishAndCloseFile() {
List<FileConnectorException> closeErrors = new ArrayList<>();
this.beingWrittenWriter.forEach(
(k, v) -> {
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/TextWriteStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/TextWriteStrategy.java
index df3ab9c4f1..fe9cbbb3d5 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/TextWriteStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/TextWriteStrategy.java
@@ -101,7 +101,7 @@ public class TextWriteStrategy extends
AbstractWriteStrategy<FSDataOutputStream>
}
@Override
- public void write(@NonNull SeaTunnelRow seaTunnelRow) {
+ public synchronized void write(@NonNull SeaTunnelRow seaTunnelRow) {
super.write(seaTunnelRow);
String filePath = getOrCreateFilePathBeingWritten(seaTunnelRow);
FSDataOutputStream fsDataOutputStream =
getOrCreateOutputStream(filePath);
@@ -118,7 +118,7 @@ public class TextWriteStrategy extends
AbstractWriteStrategy<FSDataOutputStream>
}
@Override
- public void finishAndCloseFile() {
+ public synchronized void finishAndCloseFile() {
beingWrittenOutputStream.forEach(
(key, value) -> {
try {
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/XmlWriteStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/XmlWriteStrategy.java
index adcc57b3dc..b37cc11652 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/XmlWriteStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/XmlWriteStrategy.java
@@ -45,7 +45,7 @@ public class XmlWriteStrategy extends
AbstractWriteStrategy<XmlWriter> {
}
@Override
- public void write(SeaTunnelRow seaTunnelRow) throws FileConnectorException
{
+ public synchronized void write(SeaTunnelRow seaTunnelRow) throws
FileConnectorException {
super.write(seaTunnelRow);
String filePath = getOrCreateFilePathBeingWritten(seaTunnelRow);
XmlWriter xmlDocWriter = getOrCreateOutputStream(filePath);
@@ -53,7 +53,7 @@ public class XmlWriteStrategy extends
AbstractWriteStrategy<XmlWriter> {
}
@Override
- public void finishAndCloseFile() {
+ public synchronized void finishAndCloseFile() {
this.beingWrittenWriter.forEach(
(k, v) -> {
try {
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/AbstractWriteStrategyTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/AbstractWriteStrategyTest.java
index 93d1c588ed..06ef8451a5 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/AbstractWriteStrategyTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/AbstractWriteStrategyTest.java
@@ -17,12 +17,22 @@
package org.apache.seatunnel.connectors.seatunnel.file.sink.writer;
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import org.apache.seatunnel.connectors.seatunnel.file.config.CompressFormat;
import
org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSinkOptions;
+import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat;
import
org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
import
org.apache.seatunnel.connectors.seatunnel.file.hadoop.HadoopFileSystemProxy;
import
org.apache.seatunnel.connectors.seatunnel.file.sink.config.FileSinkConfig;
+import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.junit.jupiter.api.Assertions;
@@ -31,7 +41,14 @@ import org.mockito.Mockito;
import java.io.File;
import java.io.IOException;
+import java.io.OutputStream;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
class AbstractWriteStrategyTest {
@@ -83,6 +100,147 @@ class AbstractWriteStrategyTest {
Mockito.verify(fs,
Mockito.never()).deleteEmptyDirectory(Mockito.anyString());
}
+ @Test
+ void shouldNotCloseTextOutputStreamWhileWriteIsInProgress() throws
Exception {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"name"}, new SeaTunnelDataType[]
{BasicType.STRING_TYPE});
+ BlockingOutputStream blockingOutputStream = new BlockingOutputStream();
+ FSDataOutputStream outputStream = new
FSDataOutputStream(blockingOutputStream, null);
+ HadoopFileSystemProxy fs = Mockito.mock(HadoopFileSystemProxy.class);
+
Mockito.when(fs.getOutputStream(Mockito.anyString())).thenReturn(outputStream);
+
+ TestTextWriteStrategy writeStrategy =
+ new TestTextWriteStrategy(newTextFileSinkConfig(rowType));
+ writeStrategy.setFileSystemProxy(fs);
+ writeStrategy.setCatalogTable(CatalogTableUtil.getCatalogTable("test",
rowType));
+ writeStrategy.setTransactionContext(JOB_ID, UUID_PREFIX);
+ writeStrategy.beginTransaction(1L);
+
+ CompletableFuture<Void> writeFuture =
+ CompletableFuture.runAsync(
+ () -> writeStrategy.write(new SeaTunnelRow(new
Object[] {"alice"})));
+ Assertions.assertTrue(blockingOutputStream.writeStarted.await(5,
TimeUnit.SECONDS));
+
+ CompletableFuture<Void> closeFuture =
+ CompletableFuture.runAsync(writeStrategy::finishAndCloseFile);
+ Assertions.assertFalse(
+ blockingOutputStream.closeCalled.await(200,
TimeUnit.MILLISECONDS),
+ "finishAndCloseFile must wait for the active write to finish
before closing files");
+
+ blockingOutputStream.releaseWrite.countDown();
+ writeFuture.get(5, TimeUnit.SECONDS);
+ closeFuture.get(5, TimeUnit.SECONDS);
+ Assertions.assertTrue(blockingOutputStream.closed.get());
+ }
+
+ @Test
+ void shouldNotAbortPrepareWhileWriteIsInProgress() throws Exception {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"name"}, new SeaTunnelDataType[]
{BasicType.STRING_TYPE});
+ BlockingOutputStream blockingOutputStream = new BlockingOutputStream();
+ FSDataOutputStream outputStream = new
FSDataOutputStream(blockingOutputStream, null);
+ HadoopFileSystemProxy fs = Mockito.mock(HadoopFileSystemProxy.class);
+
Mockito.when(fs.getOutputStream(Mockito.anyString())).thenReturn(outputStream);
+ CountDownLatch deleteCalled = new CountDownLatch(1);
+ Mockito.doAnswer(
+ invocation -> {
+ deleteCalled.countDown();
+ return null;
+ })
+ .when(fs)
+ .deleteFile(Mockito.anyString());
+
+ TestTextWriteStrategy writeStrategy =
+ new TestTextWriteStrategy(newTextFileSinkConfig(rowType));
+ writeStrategy.setFileSystemProxy(fs);
+ writeStrategy.setCatalogTable(CatalogTableUtil.getCatalogTable("test",
rowType));
+ writeStrategy.setTransactionContext(JOB_ID, UUID_PREFIX);
+ writeStrategy.beginTransaction(1L);
+
+ CompletableFuture<Void> writeFuture =
+ CompletableFuture.runAsync(
+ () -> writeStrategy.write(new SeaTunnelRow(new
Object[] {"alice"})));
+ Assertions.assertTrue(blockingOutputStream.writeStarted.await(5,
TimeUnit.SECONDS));
+
+ CompletableFuture<Void> abortFuture =
+ CompletableFuture.runAsync(() ->
writeStrategy.abortPrepare(TRANSACTION_ID));
+ Assertions.assertFalse(
+ deleteCalled.await(200, TimeUnit.MILLISECONDS),
+ "abortPrepare must wait for the active write to finish before
deleting the"
+ + " transaction directory");
+
+ blockingOutputStream.releaseWrite.countDown();
+ writeFuture.get(5, TimeUnit.SECONDS);
+ abortFuture.get(5, TimeUnit.SECONDS);
+ Assertions.assertTrue(deleteCalled.await(5, TimeUnit.SECONDS));
+ }
+
+ @Test
+ void shouldNotCloseOrcWriterWhileWriteIsInProgress() throws Exception {
+ assertCloseWaitsForColumnarWrite(Mockito.spy(new
OrcWriteStrategy(newFileSinkConfig())));
+ }
+
+ @Test
+ void shouldNotCloseParquetWriterWhileWriteIsInProgress() throws Exception {
+ assertCloseWaitsForColumnarWrite(
+ Mockito.spy(new ParquetWriteStrategy(newFileSinkConfig())));
+ }
+
+ private static void
assertCloseWaitsForColumnarWrite(AbstractWriteStrategy<?> strategy)
+ throws Exception {
+ CountDownLatch writeStarted = new CountDownLatch(1);
+ CountDownLatch releaseWrite = new CountDownLatch(1);
+ CountDownLatch closeStarted = new CountDownLatch(1);
+ CountDownLatch closeFinished = new CountDownLatch(1);
+ RuntimeException stopWrite = new RuntimeException("stop after writer
lock probe");
+ Mockito.doAnswer(
+ invocation -> {
+ writeStarted.countDown();
+ if (!releaseWrite.await(5, TimeUnit.SECONDS)) {
+ throw new AssertionError("write was not
released");
+ }
+ throw stopWrite;
+ })
+ .when(strategy)
+ .getOrCreateFilePathBeingWritten(Mockito.any());
+ CompletableFuture<Void> write =
+ CompletableFuture.runAsync(
+ () -> {
+ RuntimeException failure =
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () ->
+ strategy.write(
+ new SeaTunnelRow(
+ new
Object[] {"row"})));
+ Assertions.assertSame(stopWrite, failure);
+ });
+ CompletableFuture<Void> close = null;
+ try {
+ Assertions.assertTrue(writeStarted.await(5, TimeUnit.SECONDS));
+ close =
+ CompletableFuture.runAsync(
+ () -> {
+ closeStarted.countDown();
+ strategy.finishAndCloseFile();
+ closeFinished.countDown();
+ });
+ Assertions.assertTrue(closeStarted.await(5, TimeUnit.SECONDS));
+ Assertions.assertFalse(
+ closeFinished.await(200, TimeUnit.MILLISECONDS),
+ "close must wait for the complete columnar write, not only
super.write");
+ } finally {
+ releaseWrite.countDown();
+ write.get(5, TimeUnit.SECONDS);
+ if (close != null) {
+ close.get(5, TimeUnit.SECONDS);
+ }
+ }
+ Assertions.assertEquals(0, closeFinished.getCount());
+ }
+
private static TestWriteStrategy
newTestWriteStrategy(HadoopFileSystemProxy fs) {
TestWriteStrategy writeStrategy = new
TestWriteStrategy(newFileSinkConfig());
writeStrategy.setFileSystemProxy(fs);
@@ -100,6 +258,15 @@ class AbstractWriteStrategyTest {
return fileSinkConfig;
}
+ private static FileSinkConfig newTextFileSinkConfig(SeaTunnelRowType
rowType) {
+ Map<String, Object> writeConfig = new HashMap<>();
+ writeConfig.put("tmp_path", TMP_PATH);
+ writeConfig.put("path", "/tmp/seatunnel-target");
+ writeConfig.put("file_format_type", FileFormat.TEXT.name());
+ return new FileSinkConfig(
+
ReadonlyConfig.fromConfig(ConfigFactory.parseMap(writeConfig)), rowType);
+ }
+
private static TransactionPath newTransactionPath() {
String transactionDir =
String.join(
@@ -149,4 +316,56 @@ class AbstractWriteStrategyTest {
@Override
public void finishAndCloseFile() {}
}
+
+ private static class TestTextWriteStrategy extends TextWriteStrategy {
+
+ private TestTextWriteStrategy(FileSinkConfig fileSinkConfig) {
+ super(fileSinkConfig);
+ }
+
+ private void setFileSystemProxy(HadoopFileSystemProxy proxy) {
+ this.hadoopFileSystemProxy = proxy;
+ }
+
+ private void setTransactionContext(String jobId, String uuidPrefix) {
+ this.jobId = jobId;
+ this.uuidPrefix = uuidPrefix;
+ }
+ }
+
+ private static class BlockingOutputStream extends OutputStream {
+ private final CountDownLatch writeStarted = new CountDownLatch(1);
+ private final CountDownLatch releaseWrite = new CountDownLatch(1);
+ private final CountDownLatch closeCalled = new CountDownLatch(1);
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+
+ @Override
+ public void write(int b) throws IOException {
+ blockWrite();
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException {
+ blockWrite();
+ }
+
+ private void blockWrite() throws IOException {
+ writeStarted.countDown();
+ try {
+ releaseWrite.await(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException(e);
+ }
+ if (closed.get()) {
+ throw new IOException("stream was closed");
+ }
+ }
+
+ @Override
+ public void close() {
+ closed.set(true);
+ closeCalled.countDown();
+ }
+ }
}
