From 96b2ce4da920a635e195372509bc7efcddf88cee Mon Sep 17 00:00:00 2001 From: Zihan Dai <99155080+PDGGK@users.noreply.github.com> Date: Sat, 15 Aug 2026 04:20:14 +1000 Subject: [PATCH] [core] Close every writer and shut both executors down when one close fails AbstractFileStoreWrite#close walked the writers map with a plain loop: for (Map> bucketWriters : writers.values()) { for (WriterContainer writerContainer : bucketWriters.values()) { writerContainer.writer.close(); ... } } writers.clear(); if (lazyCompactExecutor != null && closeCompactExecutorWhenLeaving) { lazyCompactExecutor.shutdownNow(); } if (lazyPrimaryKeyIndexExecutor != null) { lazyPrimaryKeyIndexExecutor.shutdownNow(); } if (compactionMetrics != null) { compactionMetrics.close(); } RecordWriter#close is the only call in there that can throw -- BucketedPrimaryKeyIndexMaintainer#close and CompactionMetrics#close are both declared void with no checked exception. So one writer throwing takes out everything after it: every remaining writer in the map, the writers.clear(), both executor shutdowns, and the metrics close. There is one writer per bucket per partition, so a single bad writer can strand a large number of them, and the two thread pools stay alive for the rest of the process. The writers now go through IOUtils.closeAll, which closes all of them and rethrows the first failure with the others attached as suppressed. The tail moves into a finally so it runs whatever the writers did; none of those four calls throws, so the writer failure is never replaced by one of them. Same shape as SortMergeReaderWithMinHeap#close (#9163) and the lookup-store chain (#9172), and it uses the helper paimon already has for it rather than adding another. Four unit tests. Restoring the plain loop fails three of them -- the later writers are not closed, the suppressed failure is lost, and the map is left populated. The fourth covers the path where nothing throws and passes either way, so the first three are not trivially red. --- .../operation/AbstractFileStoreWrite.java | 36 ++- .../AbstractFileStoreWriteCloseTest.java | 221 ++++++++++++++++++ 2 files changed, 246 insertions(+), 11 deletions(-) create mode 100644 paimon-core/src/test/java/org/apache/paimon/operation/AbstractFileStoreWriteCloseTest.java diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java index 2ee11440dc8b..0c0738c6173e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java @@ -43,6 +43,7 @@ import org.apache.paimon.types.RowType; import org.apache.paimon.utils.CommitIncrement; import org.apache.paimon.utils.ExecutorThreadFactory; +import org.apache.paimon.utils.IOUtils; import org.apache.paimon.utils.RecordWriter; import org.apache.paimon.utils.RowDataToObjectArrayConverter; import org.apache.paimon.utils.SnapshotManager; @@ -362,23 +363,36 @@ Function, Boolean> createNoConflictAwareWriterCleanChecker() @Override public void close() throws Exception { + List writerCloseables = new ArrayList<>(); for (Map> bucketWriters : writers.values()) { for (WriterContainer writerContainer : bucketWriters.values()) { - writerContainer.writer.close(); + writerCloseables.add(writerContainer.writer::close); if (writerContainer.primaryKeyIndexMaintainer != null) { - writerContainer.primaryKeyIndexMaintainer.close(); + writerCloseables.add(writerContainer.primaryKeyIndexMaintainer::close); } } } - writers.clear(); - if (lazyCompactExecutor != null && closeCompactExecutorWhenLeaving) { - lazyCompactExecutor.shutdownNow(); - } - if (lazyPrimaryKeyIndexExecutor != null) { - lazyPrimaryKeyIndexExecutor.shutdownNow(); - } - if (compactionMetrics != null) { - compactionMetrics.close(); + + try { + // There is one writer per bucket per partition, each holding its own files and + // buffers. Closing them in a plain loop meant the first failure abandoned every + // writer behind it; closeAll runs all of them and rethrows the first failure with + // the rest attached to it as suppressed. + IOUtils.closeAll(writerCloseables); + } finally { + // These have to run whatever the writers did. Previously a single failing writer + // also left both thread pools running for the life of the process. None of the + // calls below throws, so the writer failure is never replaced by one of them. + writers.clear(); + if (lazyCompactExecutor != null && closeCompactExecutorWhenLeaving) { + lazyCompactExecutor.shutdownNow(); + } + if (lazyPrimaryKeyIndexExecutor != null) { + lazyPrimaryKeyIndexExecutor.shutdownNow(); + } + if (compactionMetrics != null) { + compactionMetrics.close(); + } } } diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/AbstractFileStoreWriteCloseTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/AbstractFileStoreWriteCloseTest.java new file mode 100644 index 000000000000..8c4ac1252f7c --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/operation/AbstractFileStoreWriteCloseTest.java @@ -0,0 +1,221 @@ +/* + * 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.paimon.operation; + +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.FileSystemCatalog; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryRowWriter; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.utils.CommitIncrement; +import org.apache.paimon.utils.RecordWriter; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link AbstractFileStoreWrite#close()}. */ +public class AbstractFileStoreWriteCloseTest { + + @TempDir java.nio.file.Path tempDir; + + @Test + public void testCloseReleasesEveryWriterWhenAnEarlierOneThrows() throws Exception { + BucketedAppendFileStoreWrite write = newWrite(); + + // Three writers in one partition. The first throws, so the plain loop this replaces + // stopped there and the two behind it were never closed. + RecordingWriter failing = new RecordingWriter("bucket-0", true); + RecordingWriter second = new RecordingWriter("bucket-1", false); + RecordingWriter third = new RecordingWriter("bucket-2", false); + putWriters(write, failing, second, third); + + assertThatThrownBy(write::close).hasMessage("close failed in bucket-0"); + + assertThat(failing.closed).as("the failing writer").isTrue(); + assertThat(second.closed).as("the writer after the failure").isTrue(); + assertThat(third.closed).as("the last writer").isTrue(); + } + + @Test + public void testLaterFailuresRideAlongInsteadOfBeingDropped() throws Exception { + BucketedAppendFileStoreWrite write = newWrite(); + + putWriters( + write, + new RecordingWriter("bucket-0", true), + new RecordingWriter("bucket-1", true)); + + assertThatThrownBy(write::close) + .hasMessage("close failed in bucket-0") + .satisfies( + thrown -> + assertThat(thrown.getSuppressed()) + .as("the second failure") + .hasSize(1) + .allSatisfy( + s -> + assertThat(s) + .hasMessage( + "close failed in bucket-1"))); + } + + @Test + public void testWriterMapIsClearedEvenWhenAWriterThrows() throws Exception { + BucketedAppendFileStoreWrite write = newWrite(); + putWriters(write, new RecordingWriter("bucket-0", true)); + + assertThatThrownBy(write::close).hasMessage("close failed in bucket-0"); + + // writers.clear() sits after the loop, together with the two executor shutdowns, so a + // throwing writer used to skip all of them. + assertThat(write.writers()).as("writers map after a failing close").isEmpty(); + } + + @Test + public void testCloseSucceedsWhenNoWriterThrows() throws Exception { + BucketedAppendFileStoreWrite write = newWrite(); + RecordingWriter a = new RecordingWriter("bucket-0", false); + RecordingWriter b = new RecordingWriter("bucket-1", false); + putWriters(write, a, b); + + write.close(); + + assertThat(a.closed).isTrue(); + assertThat(b.closed).isTrue(); + assertThat(write.writers()).isEmpty(); + } + + private void putWriters(BucketedAppendFileStoreWrite write, RecordingWriter... writers) { + HashMap> bucketWriters = + new HashMap<>(); + for (int i = 0; i < writers.length; i++) { + bucketWriters.put( + i, + new AbstractFileStoreWrite.WriterContainer<>( + writers[i], 1, null, null, null, null)); + } + write.writers().put(partition(0), bucketWriters); + } + + private BucketedAppendFileStoreWrite newWrite() throws Exception { + return (BucketedAppendFileStoreWrite) createFileStoreTable().store().newWrite("ss"); + } + + private static BinaryRow partition(int i) { + BinaryRow binaryRow = new BinaryRow(1); + BinaryRowWriter writer = new BinaryRowWriter(binaryRow); + writer.writeInt(0, i); + writer.complete(); + return binaryRow; + } + + private FileStoreTable createFileStoreTable() throws Exception { + Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), new Path(tempDir.toString())); + Schema schema = + Schema.newBuilder() + .column("f0", DataTypes.INT()) + .column("f1", DataTypes.INT()) + .column("f2", DataTypes.INT()) + .partitionKeys("f0") + .option("bucket", "100") + .option("bucket-key", "f1") + .build(); + Identifier identifier = Identifier.create("default", "test"); + catalog.createDatabase("default", false); + catalog.createTable(identifier, schema, false); + return (FileStoreTable) catalog.getTable(identifier); + } + + /** + * Records whether it was closed, and optionally throws when it is. {@code close()} is the only + * method {@link AbstractFileStoreWrite#close()} reaches. + */ + private static class RecordingWriter implements RecordWriter { + + private final String name; + private final boolean throwOnClose; + private boolean closed = false; + + private RecordingWriter(String name, boolean throwOnClose) { + this.name = name; + this.throwOnClose = throwOnClose; + } + + @Override + public void close() throws Exception { + closed = true; + if (throwOnClose) { + throw new Exception("close failed in " + name); + } + } + + @Override + public void write(InternalRow record) { + throw new UnsupportedOperationException(); + } + + @Override + public void compact(boolean fullCompaction) { + throw new UnsupportedOperationException(); + } + + @Override + public void addNewFiles(List files) { + throw new UnsupportedOperationException(); + } + + @Override + public Collection dataFiles() { + return Collections.emptyList(); + } + + @Override + public long maxSequenceNumber() { + return 0; + } + + @Override + public CommitIncrement prepareCommit(boolean waitCompaction) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean compactNotCompleted() { + return false; + } + + @Override + public void sync() {} + } +}