This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new d781a1e19f [core] Continue FileStoreWrite cleanup after writer close
failures (#10288)
d781a1e19f is described below
commit d781a1e19fc983b87c85e40d543c6afca43e79ce
Author: ZIHAN DAI <[email protected]>
AuthorDate: Fri Oct 2 11:45:23 2026 +1000
[core] Continue FileStoreWrite cleanup after writer close failures (#10288)
---
.../paimon/operation/BaseAppendFileStoreWrite.java | 8 +-
.../paimon/operation/KeyValueFileStoreWrite.java | 4 +-
.../paimon/operation/MemoryFileStoreWrite.java | 8 +-
.../operation/AbstractFileStoreWriteCloseTest.java | 108 ++++++++++++++++++++-
4 files changed, 115 insertions(+), 13 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
index 2e734552bd..07d15637fe 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
@@ -46,6 +46,7 @@ import org.apache.paimon.utils.CommitIncrement;
import org.apache.paimon.utils.ExceptionUtils;
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.IOExceptionSupplier;
+import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.LongCounter;
import org.apache.paimon.utils.MutableObjectIterator;
import org.apache.paimon.utils.RecordWriter;
@@ -256,9 +257,10 @@ public abstract class BaseAppendFileStoreWrite extends
MemoryFileStoreWrite<Inte
@Override
public void close() throws Exception {
- super.close();
- if (blobFetchMetrics != null) {
- blobFetchMetrics.close();
+ if (blobFetchMetrics == null) {
+ super.close();
+ } else {
+ IOUtils.closeAll(super::close, blobFetchMetrics::close);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
index a81516b8e7..b685a152c7 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
@@ -49,6 +49,7 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.CommitIncrement;
import org.apache.paimon.utils.FieldsComparator;
import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.UserDefinedSeqComparator;
@@ -258,7 +259,6 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
@Override
public void close() throws Exception {
- super.close();
- compactManagerFactory.close();
+ IOUtils.closeAll(super::close, compactManagerFactory::close);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/MemoryFileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/MemoryFileStoreWrite.java
index 96a6d6b483..f6b67fad4d 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/MemoryFileStoreWrite.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/MemoryFileStoreWrite.java
@@ -30,6 +30,7 @@ import org.apache.paimon.metrics.MetricRegistry;
import org.apache.paimon.operation.metrics.WriterBufferMetric;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.RecordWriter;
import org.apache.paimon.utils.SnapshotManager;
@@ -161,9 +162,10 @@ public abstract class MemoryFileStoreWrite<T> extends
AbstractFileStoreWrite<T>
@Override
public void close() throws Exception {
- super.close();
- if (this.writerBufferMetric != null) {
- this.writerBufferMetric.close();
+ if (writerBufferMetric == null) {
+ super.close();
+ } else {
+ IOUtils.closeAll(super::close, writerBufferMetric::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
index 8c4ac1252f..95125d6908 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/AbstractFileStoreWriteCloseTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/AbstractFileStoreWriteCloseTest.java
@@ -18,6 +18,7 @@
package org.apache.paimon.operation;
+import org.apache.paimon.KeyValue;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.FileSystemCatalog;
import org.apache.paimon.catalog.Identifier;
@@ -27,6 +28,11 @@ 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.mergetree.compact.KvCompactionManagerFactory;
+import org.apache.paimon.metrics.MetricGroupImpl;
+import org.apache.paimon.metrics.MetricRegistry;
+import org.apache.paimon.operation.metrics.BlobFetchMetrics;
+import org.apache.paimon.operation.metrics.WriterBufferMetric;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.DataTypes;
@@ -36,10 +42,13 @@ import org.apache.paimon.utils.RecordWriter;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import java.lang.reflect.Field;
+import java.lang.reflect.Proxy;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -116,6 +125,88 @@ public class AbstractFileStoreWriteCloseTest {
assertThat(write.writers()).isEmpty();
}
+ @Test
+ public void testAppendCloseRunsBlobMetricCleanupAfterWriterFailure()
throws Exception {
+ BucketedAppendFileStoreWrite write = newWrite();
+ RecordingWriter failing = new RecordingWriter("bucket-0", true);
+ putWriters(write, failing);
+
+ AtomicBoolean blobMetricClosed = new AtomicBoolean();
+ MetricRegistry metricRegistry =
+ (groupName, variables) ->
+ new MetricGroupImpl(groupName, variables) {
+ @Override
+ public void close() {
+ if
(BlobFetchMetrics.GROUP_NAME.equals(groupName)) {
+ blobMetricClosed.set(true);
+ }
+ }
+ };
+ Field blobMetricField =
BaseAppendFileStoreWrite.class.getDeclaredField("blobFetchMetrics");
+ blobMetricField.setAccessible(true);
+ blobMetricField.set(write, new BlobFetchMetrics(metricRegistry,
"test"));
+
+ assertThatThrownBy(write::close).hasMessage("close failed in
bucket-0");
+ assertThat(blobMetricClosed).isTrue();
+ }
+
+ @Test
+ public void testKeyValueCloseRunsOuterCleanupAfterWriterFailure() throws
Exception {
+ KeyValueFileStoreWrite write = newKeyValueWrite();
+ AtomicBoolean metricClosed = new AtomicBoolean();
+ MetricRegistry metricRegistry =
+ (groupName, variables) ->
+ new MetricGroupImpl(groupName, variables) {
+ @Override
+ public void close() {
+ if
(WriterBufferMetric.GROUP_NAME.equals(groupName)) {
+ metricClosed.set(true);
+ }
+ }
+ };
+ write.withMetricRegistry(metricRegistry);
+
+ @SuppressWarnings("unchecked")
+ RecordWriter<KeyValue> failingWriter =
+ (RecordWriter<KeyValue>)
+ Proxy.newProxyInstance(
+ RecordWriter.class.getClassLoader(),
+ new Class<?>[] {RecordWriter.class},
+ (proxy, method, args) -> {
+ if ("close".equals(method.getName())) {
+ throw new Exception("writer close
failed");
+ }
+ return null;
+ });
+ HashMap<Integer, AbstractFileStoreWrite.WriterContainer<KeyValue>>
bucketWriters =
+ new HashMap<>();
+ bucketWriters.put(
+ 0,
+ new AbstractFileStoreWrite.WriterContainer<>(
+ failingWriter, 1, null, null, null, null));
+ write.writers().put(partition(0), bucketWriters);
+
+ AtomicBoolean factoryClosed = new AtomicBoolean();
+ KvCompactionManagerFactory compactManagerFactory =
+ (KvCompactionManagerFactory)
+ Proxy.newProxyInstance(
+
KvCompactionManagerFactory.class.getClassLoader(),
+ new Class<?>[]
{KvCompactionManagerFactory.class},
+ (proxy, method, args) -> {
+ if ("close".equals(method.getName())) {
+ factoryClosed.set(true);
+ }
+ return null;
+ });
+ Field factoryField =
KeyValueFileStoreWrite.class.getDeclaredField("compactManagerFactory");
+ factoryField.setAccessible(true);
+ factoryField.set(write, compactManagerFactory);
+
+ assertThatThrownBy(write::close).hasMessage("writer close failed");
+ assertThat(metricClosed).isTrue();
+ assertThat(factoryClosed).isTrue();
+ }
+
private void putWriters(BucketedAppendFileStoreWrite write,
RecordingWriter... writers) {
HashMap<Integer, AbstractFileStoreWrite.WriterContainer<InternalRow>>
bucketWriters =
new HashMap<>();
@@ -129,7 +220,11 @@ public class AbstractFileStoreWriteCloseTest {
}
private BucketedAppendFileStoreWrite newWrite() throws Exception {
- return (BucketedAppendFileStoreWrite)
createFileStoreTable().store().newWrite("ss");
+ return (BucketedAppendFileStoreWrite)
createFileStoreTable(false).store().newWrite("ss");
+ }
+
+ private KeyValueFileStoreWrite newKeyValueWrite() throws Exception {
+ return (KeyValueFileStoreWrite)
createFileStoreTable(true).store().newWrite("ss");
}
private static BinaryRow partition(int i) {
@@ -140,17 +235,20 @@ public class AbstractFileStoreWriteCloseTest {
return binaryRow;
}
- private FileStoreTable createFileStoreTable() throws Exception {
+ private FileStoreTable createFileStoreTable(boolean primaryKey) throws
Exception {
Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), new
Path(tempDir.toString()));
- Schema schema =
+ Schema.Builder schemaBuilder =
Schema.newBuilder()
.column("f0", DataTypes.INT())
.column("f1", DataTypes.INT())
.column("f2", DataTypes.INT())
.partitionKeys("f0")
.option("bucket", "100")
- .option("bucket-key", "f1")
- .build();
+ .option("bucket-key", "f1");
+ if (primaryKey) {
+ schemaBuilder.primaryKey("f0", "f1");
+ }
+ Schema schema = schemaBuilder.build();
Identifier identifier = Identifier.create("default", "test");
catalog.createDatabase("default", false);
catalog.createTable(identifier, schema, false);
