This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new a833f881ebca fix(common): clean up LSM spill files when source
iterator close fails (#20023)
a833f881ebca is described below
commit a833f881ebca528b7d47a43efb351d05f45d68be
Author: Shuo Cheng <[email protected]>
AuthorDate: Wed Sep 23 12:16:27 2026 +0800
fix(common): clean up LSM spill files when source iterator close fails
(#20023)
* fix(common): clean up LSM spill files when source iterator close fails
---
.../table/read/lsm/SpillableLsmRecordIterator.java | 12 +-
.../read/lsm/TestSpillableLsmRecordIterator.java | 121 ++++++++++++++++-----
2 files changed, 104 insertions(+), 29 deletions(-)
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/lsm/SpillableLsmRecordIterator.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/lsm/SpillableLsmRecordIterator.java
index 8702562024ed..9616c03bde67 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/lsm/SpillableLsmRecordIterator.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/lsm/SpillableLsmRecordIterator.java
@@ -22,6 +22,7 @@ package org.apache.hudi.common.table.read.lsm;
import org.apache.hudi.common.engine.RecordContext;
import org.apache.hudi.common.serialization.CustomSerializer;
import org.apache.hudi.common.table.read.BufferedRecord;
+import org.apache.hudi.common.util.CloseableUtils;
import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.exception.HoodieIOException;
@@ -72,7 +73,7 @@ class SpillableLsmRecordIterator<T> implements
ClosableIterator<BufferedRecord<T
} catch (IOException e) {
spillFailure = e;
throw new HoodieIOException("Failed to spill LSM input iterator", e);
- } catch (RuntimeException e) {
+ } catch (Throwable e) {
spillFailure = e;
throw e;
} finally {
@@ -89,8 +90,8 @@ class SpillableLsmRecordIterator<T> implements
ClosableIterator<BufferedRecord<T
outputStream.write(bytes);
count++;
}
- } catch (IOException | RuntimeException e) {
- deleteSpillFile();
+ } catch (Throwable e) {
+ CloseableUtils.closeSuppressing(this::deleteSpillFile, e);
throw e;
}
return count;
@@ -162,10 +163,13 @@ class SpillableLsmRecordIterator<T> implements
ClosableIterator<BufferedRecord<T
Throwable spillFailure) {
try {
sourceIterator.close();
- } catch (RuntimeException e) {
+ } catch (Throwable e) {
if (spillFailure != null) {
spillFailure.addSuppressed(e);
} else {
+ // Closing the source iterator failed, so construction cannot complete
and the outer reader
+ // cannot call close() on this spill iterator. Delete its spill file
here to avoid leaking it.
+ CloseableUtils.closeSuppressing(this::deleteSpillFile, e);
throw e;
}
}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/lsm/TestSpillableLsmRecordIterator.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/lsm/TestSpillableLsmRecordIterator.java
index 2b33d1b9d1b3..591bec0c13ea 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/lsm/TestSpillableLsmRecordIterator.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/lsm/TestSpillableLsmRecordIterator.java
@@ -26,11 +26,16 @@ import org.apache.hudi.exception.HoodieIOException;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
+import java.nio.file.DirectoryNotEmptyException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.stream.Stream;
@@ -39,6 +44,10 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
class TestSpillableLsmRecordIterator {
@@ -67,10 +76,11 @@ class TestSpillableLsmRecordIterator {
assertEquals(0, spillFileCount());
}
- @Test
- void testSpillFailurePreservesSourceCloseFailureAsSuppressed() throws
IOException {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testSpillFailurePreservesSourceCloseFailureAsSuppressed(boolean
closeError) throws IOException {
Path spillBaseFile = Files.createTempFile(tempDir, "spill-base", ".tmp");
- RuntimeException closeFailure = new RuntimeException("source close
failed");
+ Throwable closeFailure = sourceCloseFailure(closeError);
HoodieIOException exception = assertThrows(HoodieIOException.class, () ->
new SpillableLsmRecordIterator<>(
closeFailingIterator(closeFailure), new DefaultSerializer<>(), null,
spillBaseFile.toString()));
@@ -106,36 +116,97 @@ class TestSpillableLsmRecordIterator {
iterator.close();
}
- @Test
- void testSuccessfulSpillPropagatesSourceCloseFailure() {
- RuntimeException closeFailure = new RuntimeException("source close
failed");
+ @ParameterizedTest
+ @CsvSource({"false, false", "false, true", "true, false", "true, true"})
+ void testSuccessfulSpillCleansUpOnSourceCloseFailure(boolean empty, boolean
closeError) throws IOException {
+ Throwable closeFailure = sourceCloseFailure(closeError);
+ List<BufferedRecord<String>> records = empty ? Collections.emptyList()
+ : Collections.singletonList(new BufferedRecord<>("key", 1, null, null,
null));
+ ClosableIterator<BufferedRecord<String>> sourceIterator =
spy(ClosableIterator.wrap(records.iterator()));
+ doThrow(closeFailure).when(sourceIterator).close();
+
+ assertSame(closeFailure, assertThrows(closeFailure.getClass(), () -> new
SpillableLsmRecordIterator<>(
+ sourceIterator, new DefaultSerializer<>(), null, tempDir.toString())));
+ assertEquals(0, spillFileCount());
+ }
- assertSame(closeFailure, assertThrows(RuntimeException.class, () -> new
SpillableLsmRecordIterator<>(
- closeFailingIterator(closeFailure), new DefaultSerializer<>(), null,
tempDir.toString())));
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testSourceCloseFailurePreservesSpillCleanupFailureAsSuppressed(boolean
closeError) {
+ Throwable closeFailure = sourceCloseFailure(closeError);
+ ClosableIterator<BufferedRecord<String>> sourceIterator =
closeFailingIterator(closeFailure);
+ doAnswer(invocation -> {
+ replaceSpillFileWithNonEmptyDirectory();
+ throw closeFailure;
+ }).when(sourceIterator).close();
+
+ Throwable exception = assertThrows(closeFailure.getClass(), () -> new
SpillableLsmRecordIterator<>(
+ sourceIterator, new DefaultSerializer<>(), null, tempDir.toString()));
+
+ assertSame(closeFailure, exception);
+ assertEquals(1, exception.getSuppressed().length);
+ assertTrue(exception.getSuppressed()[0] instanceof HoodieIOException);
+ assertTrue(exception.getSuppressed()[0].getCause() instanceof
DirectoryNotEmptyException);
}
- private long spillFileCount() throws IOException {
+ @ParameterizedTest
+ @CsvSource({"false, false", "false, true", "true, false", "true, true"})
+ void testSpillErrorPreservesCleanupAndSourceCloseFailures(boolean
cleanupFails, boolean closeFails) throws IOException {
+ AssertionError spillFailure = new AssertionError("spill failed");
+ AssertionError closeFailure = new AssertionError("source close failed");
+ ClosableIterator<BufferedRecord<String>> sourceIterator =
spy(ClosableIterator.wrap(Collections.emptyIterator()));
+ doAnswer(invocation -> {
+ assertEquals(1, spillFileCount());
+ if (cleanupFails) {
+ replaceSpillFileWithNonEmptyDirectory();
+ }
+ throw spillFailure;
+ }).when(sourceIterator).hasNext();
+ if (closeFails) {
+ doThrow(closeFailure).when(sourceIterator).close();
+ }
+
+ AssertionError exception = assertThrows(AssertionError.class, () -> new
SpillableLsmRecordIterator<>(
+ sourceIterator, new DefaultSerializer<>(), null, tempDir.toString()));
+
+ assertSame(spillFailure, exception);
+ Throwable[] suppressed = exception.getSuppressed();
+ assertEquals((cleanupFails ? 1 : 0) + (closeFails ? 1 : 0),
suppressed.length);
+ if (cleanupFails) {
+ assertTrue(suppressed[0] instanceof HoodieIOException);
+ assertTrue(suppressed[0].getCause() instanceof
DirectoryNotEmptyException);
+ }
+ if (closeFails) {
+ assertSame(closeFailure, suppressed[suppressed.length - 1]);
+ }
+ verify(sourceIterator).close();
+ assertEquals(cleanupFails ? 1 : 0, spillFileCount());
+ }
+
+ private void replaceSpillFileWithNonEmptyDirectory() throws IOException {
+ Path spillFile;
try (Stream<Path> paths = Files.list(tempDir)) {
- return paths.count();
+ spillFile = paths.findFirst().get();
}
+ // A non-empty directory makes deletion fail reliably without relying on
filesystem permissions.
+ Files.delete(spillFile);
+ Files.createDirectory(spillFile);
+ Files.createFile(spillFile.resolve("child"));
}
- private ClosableIterator<BufferedRecord<String>>
closeFailingIterator(RuntimeException closeFailure) {
- return new ClosableIterator<BufferedRecord<String>>() {
- @Override
- public boolean hasNext() {
- return false;
- }
+ private static Throwable sourceCloseFailure(boolean error) {
+ return error ? new AssertionError("source close failed") : new
RuntimeException("source close failed");
+ }
- @Override
- public BufferedRecord<String> next() {
- throw new UnsupportedOperationException();
- }
+ private long spillFileCount() throws IOException {
+ try (Stream<Path> paths = Files.list(tempDir)) {
+ return paths.count();
+ }
+ }
- @Override
- public void close() {
- throw closeFailure;
- }
- };
+ private ClosableIterator<BufferedRecord<String>>
closeFailingIterator(Throwable closeFailure) {
+ ClosableIterator<BufferedRecord<String>> sourceIterator =
spy(ClosableIterator.wrap(Collections.emptyIterator()));
+ doThrow(closeFailure).when(sourceIterator).close();
+ return sourceIterator;
}
}
