This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-11103-719c3e2b311fa6a050749984a43e6a837b9f802f in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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(); + } + } }
