From d1998a6986cfbd36735d7b379f0d6a2455033304 Mon Sep 17 00:00:00 2001 From: danny0405 Date: Mon, 3 Aug 2026 17:40:04 +0800 Subject: [PATCH 1/5] fix(flink): close CDC image spillable maps on failures --- .../table/format/cdc/CdcImageManager.java | 26 ++++++++-- .../hudi/table/format/cdc/CdcIterators.java | 15 +++--- .../table/format/cdc/TestCdcImageManager.java | 49 +++++++++++++++++++ 3 files changed, 79 insertions(+), 11 deletions(-) diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcImageManager.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcImageManager.java index 2854498c9bdab..87e760c6247d8 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcImageManager.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcImageManager.java @@ -86,7 +86,12 @@ public ExternalSpillableMap getOrLoadImages( cache.remove(oldest).close(); } ExternalSpillableMap images = loadImageRecords(maxCompactionMemoryInBytes, fileSlice); - cache.put(instant, images); + try { + cache.put(instant, images); + } catch (RuntimeException | Error e) { + closeSuppressing(images, e); + throw e; + } return images; } @@ -105,13 +110,26 @@ private ExternalSpillableMap loadImageRecords( serializer.serialize(row, new BytesArrayOutputView(baos)); imageRecordsMap.put(recordKey, baos.toByteArray()); } + } catch (IOException | RuntimeException | Error e) { + closeSuppressing(imageRecordsMap, e); + throw e; } return imageRecordsMap; } + private static void closeSuppressing( + ExternalSpillableMap imageRecordsMap, + Throwable primary) { + try { + imageRecordsMap.close(); + } catch (RuntimeException | Error closeError) { + primary.addSuppressed(closeError); + } + } + public RowData getImageRecord( String recordKey, - ExternalSpillableMap imageCache, + Map imageCache, RowKind rowKind) { byte[] bytes = imageCache.get(recordKey); ValidationUtils.checkState(bytes != null, @@ -127,7 +145,7 @@ public RowData getImageRecord( public void updateImageRecord( String recordKey, - ExternalSpillableMap imageCache, + Map imageCache, RowData row) { ByteArrayOutputStream baos = new ByteArrayOutputStream(4096); try { @@ -140,7 +158,7 @@ public void updateImageRecord( public RowData removeImageRecord( String recordKey, - ExternalSpillableMap imageCache) { + Map imageCache) { byte[] bytes = imageCache.remove(recordKey); if (bytes == null) { return null; diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java index 725873329bcc5..c1ec9a17cc5e0 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java @@ -60,7 +60,6 @@ import org.apache.hudi.storage.HoodieStorage; import org.apache.hudi.storage.StoragePath; import org.apache.hudi.table.format.FlinkReaderContextFactory; -import org.apache.hudi.table.format.FormatUtils; import org.apache.hudi.table.format.HoodieRowDataFileReader; import org.apache.hudi.table.format.InternalSchemaManager; import org.apache.hudi.table.format.mor.MergeOnReadInputSplit; @@ -80,6 +79,7 @@ import java.util.Collections; import java.util.Iterator; import java.util.List; +import java.util.Map; import java.util.function.Function; import java.util.stream.Collectors; @@ -253,7 +253,7 @@ public static class DataLogFileIterator implements ClosableIterator { private final String[] orderingFields; private final TypedProperties props; - private ExternalSpillableMap beforeImages; + private Map beforeImages; private RowData currentImage; private RowData sideImage; @@ -287,15 +287,15 @@ public DataLogFileIterator( metaClient.getTableConfig().getPartialUpdateMode()); this.logRecordIterator = logRecordIterator; this.deleteContext = new DeleteContext(props, tableSchema).withReaderSchema(tableSchema); - initImages(cdcFileSplit, writeConfig); + initImages(cdcFileSplit); } - private void initImages(HoodieCDCFileSplit fileSplit, HoodieWriteConfig writeConfig) throws IOException { + private void initImages(HoodieCDCFileSplit fileSplit) throws IOException { if (fileSplit.getBeforeFileSlice().isPresent() && !fileSplit.getBeforeFileSlice().get().isEmpty()) { this.beforeImages = imageManager.getOrLoadImages( maxCompactionMemoryInBytes, fileSplit.getBeforeFileSlice().get()); } else { - this.beforeImages = FormatUtils.spillableMap(writeConfig, maxCompactionMemoryInBytes, getClass().getSimpleName()); + this.beforeImages = Collections.emptyMap(); } } @@ -347,8 +347,9 @@ public RowData next() { @Override public void close() { - logRecordIterator.close(); - imageManager.close(); + try (CdcImageManager ignored = imageManager) { + logRecordIterator.close(); + } } @SuppressWarnings("unchecked") diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/cdc/TestCdcImageManager.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/cdc/TestCdcImageManager.java index e761a2a7945f9..cb7398420ccaa 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/cdc/TestCdcImageManager.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/cdc/TestCdcImageManager.java @@ -154,6 +154,55 @@ void testImageCacheReuseEvictionAndClose() throws IOException { } } + @Test + void testLoadClosesImageCacheWhenIteratorCreationFails() { + HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class); + when(writeConfig.getBasePath()).thenReturn("/table"); + ExternalSpillableMap imageCache = mockImageCache(); + RuntimeException failure = new RuntimeException("iterator creation failed"); + CdcImageManager imageManager = new CdcImageManager( + rowType("value"), + writeConfig, + split -> { + throw failure; + }); + + try (MockedStatic mockedFormatUtils = mockStatic(FormatUtils.class)) { + mockedFormatUtils.when(() -> FormatUtils.spillableMap( + writeConfig, 1024L, CdcImageManager.class.getSimpleName())) + .thenReturn(imageCache); + + assertSame(failure, assertThrows( + RuntimeException.class, + () -> imageManager.getOrLoadImages(1024L, fileSlice("001")))); + verify(imageCache).close(); + } + } + + @Test + void testLoadClosesIteratorAndImageCacheWhenIterationFails() { + HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class); + when(writeConfig.getBasePath()).thenReturn("/table"); + ExternalSpillableMap imageCache = mockImageCache(); + ClosableIterator iterator = mock(ClosableIterator.class); + RuntimeException failure = new RuntimeException("iteration failed"); + when(iterator.hasNext()).thenThrow(failure); + CdcImageManager imageManager = new CdcImageManager( + rowType("value"), writeConfig, split -> iterator); + + try (MockedStatic mockedFormatUtils = mockStatic(FormatUtils.class)) { + mockedFormatUtils.when(() -> FormatUtils.spillableMap( + writeConfig, 1024L, CdcImageManager.class.getSimpleName())) + .thenReturn(imageCache); + + assertSame(failure, assertThrows( + RuntimeException.class, + () -> imageManager.getOrLoadImages(1024L, fileSlice("001")))); + verify(iterator).close(); + verify(imageCache).close(); + } + } + @SuppressWarnings("unchecked") private static ExternalSpillableMap mockImageCache() { ExternalSpillableMap imageCache = mock(ExternalSpillableMap.class); From adb73870b52ed952065c19a4c2732ee1744a586c Mon Sep 17 00:00:00 2001 From: danny0405 Date: Tue, 4 Aug 2026 08:54:33 +0800 Subject: [PATCH 2/5] fix(flink): harden CDC cleanup failure paths --- .../HoodieCdcSplitReaderFunction.java | 21 ++++- .../table/format/cdc/CdcImageManager.java | 32 ++++--- .../hudi/table/format/cdc/CdcInputFormat.java | 23 +++-- .../hudi/table/format/cdc/CdcIterators.java | 18 +++- .../table/format/cdc/TestCdcImageManager.java | 87 ++++++++++++++++--- 5 files changed, 147 insertions(+), 34 deletions(-) diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieCdcSplitReaderFunction.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieCdcSplitReaderFunction.java index cd4ea053bc7d6..f865d88e51469 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieCdcSplitReaderFunction.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieCdcSplitReaderFunction.java @@ -223,10 +223,15 @@ conf, getHadoopConf(), tablePath, tableSchema, requiredSchema, String logFilePath = new Path(tablePath, fileSplit.getCdcFiles().get(0)).toString(); MergeOnReadInputSplit split = CdcIterators.singleLogFile2Split(tablePath, logFilePath, maxCompactionMemoryInBytes); ClosableIterator> recordIterator = getFileSliceHoodieRecordIterator(split); - return new CdcIterators.DataLogFileIterator( - maxCompactionMemoryInBytes, imageManager, fileSplit, tableSchema, - tableState.getRequiredRowType(), tableState.getRequiredPositions(), - recordIterator, getMetaClient(), getWriteConfig()); + try { + return new CdcIterators.DataLogFileIterator( + maxCompactionMemoryInBytes, imageManager, fileSplit, tableSchema, + tableState.getRequiredRowType(), tableState.getRequiredPositions(), + recordIterator, getMetaClient(), getWriteConfig()); + } catch (IOException | RuntimeException | Error e) { + closeSuppressing(recordIterator, e); + throw e; + } } case REPLACE_COMMIT: { return new CdcIterators.ReplaceCommitIterator( @@ -238,6 +243,14 @@ conf, getHadoopConf(), tablePath, tableSchema, requiredSchema, } } + private static void closeSuppressing(ClosableIterator iterator, Throwable primary) { + try { + iterator.close(); + } catch (RuntimeException | Error closeError) { + primary.addSuppressed(closeError); + } + } + /** Reads the full-schema before/after image for a file slice (emitDelete=false). */ private ClosableIterator getFileSliceIterator(MergeOnReadInputSplit split) { FileSlice fileSlice = buildFileSlice(split); diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcImageManager.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcImageManager.java index 87e760c6247d8..cdc72f7798f21 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcImageManager.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcImageManager.java @@ -86,12 +86,7 @@ public ExternalSpillableMap getOrLoadImages( cache.remove(oldest).close(); } ExternalSpillableMap images = loadImageRecords(maxCompactionMemoryInBytes, fileSlice); - try { - cache.put(instant, images); - } catch (RuntimeException | Error e) { - closeSuppressing(images, e); - throw e; - } + cache.put(instant, images); return images; } @@ -118,10 +113,10 @@ private ExternalSpillableMap loadImageRecords( } private static void closeSuppressing( - ExternalSpillableMap imageRecordsMap, + ExternalSpillableMap spillableMap, Throwable primary) { try { - imageRecordsMap.close(); + spillableMap.close(); } catch (RuntimeException | Error closeError) { primary.addSuppressed(closeError); } @@ -172,8 +167,25 @@ public RowData removeImageRecord( @Override public void close() { - cache.values().forEach(ExternalSpillableMap::close); - cache.clear(); + RuntimeException failure = null; + try { + for (ExternalSpillableMap spillableMap : cache.values()) { + try { + spillableMap.close(); + } catch (RuntimeException e) { + if (failure == null) { + failure = e; + } else { + failure.addSuppressed(e); + } + } + } + } finally { + cache.clear(); + } + if (failure != null) { + throw failure; + } } // ------------------------------------------------------------------------- diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcInputFormat.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcInputFormat.java index ff810bb4906cd..f7c5c9cef1832 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcInputFormat.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcInputFormat.java @@ -157,11 +157,16 @@ private ClosableIterator getRecordIterator( String logFilepath = new Path(tablePath, fileSplit.getCdcFiles().get(0)).toString(); MergeOnReadInputSplit split = CdcIterators.singleLogFile2Split(tablePath, logFilepath, maxCompactionMemoryInBytes); ClosableIterator> recordIterator = getSplitRecordIterator(split); - return new CdcIterators.DataLogFileIterator( - maxCompactionMemoryInBytes, imageManager, fileSplit, - HoodieSchema.parse(tableState.getTableSchema()), - tableState.getRequiredRowType(), tableState.getRequiredPositions(), - recordIterator, metaClient, imageManager.getWriteConfig()); + try { + return new CdcIterators.DataLogFileIterator( + maxCompactionMemoryInBytes, imageManager, fileSplit, + HoodieSchema.parse(tableState.getTableSchema()), + tableState.getRequiredRowType(), tableState.getRequiredPositions(), + recordIterator, metaClient, imageManager.getWriteConfig()); + } catch (IOException | RuntimeException | Error e) { + closeSuppressing(recordIterator, e); + throw e; + } case REPLACE_COMMIT: return new CdcIterators.ReplaceCommitIterator( tablePath, tableState.getRequiredRowType(), tableState.getRequiredPositions(), @@ -171,6 +176,14 @@ private ClosableIterator getRecordIterator( } } + private static void closeSuppressing(ClosableIterator iterator, Throwable primary) { + try { + iterator.close(); + } catch (RuntimeException | Error closeError) { + primary.addSuppressed(closeError); + } + } + /** * Get a {@link HoodieRecord} iterator using a {@link HoodieRecordReader}. * diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java index c1ec9a17cc5e0..c01d74b229f66 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java @@ -347,6 +347,9 @@ public RowData next() { @Override public void close() { + // Closing the shared image manager here is required for correctness: this iterator + // destructively updates cached before-images, and LOG_FILE slices from the same instant + // share a cache key. Closing forces the next split to reload its own before-images. try (CdcImageManager ignored = imageManager) { logRecordIterator.close(); } @@ -663,7 +666,12 @@ public BeforeImageIterator( this.maxCompactionMemoryInBytes = maxCompactionMemoryInBytes; this.projection = RowDataProjection.instance(requiredRowType, requiredPositions); this.imageManager = imageManager; - initImages(fileSplit); + try { + initImages(fileSplit); + } catch (IOException | RuntimeException | Error e) { + closeSuppressing(this, e); + throw e; + } } protected void initImages(HoodieCDCFileSplit fileSplit) throws IOException { @@ -729,6 +737,14 @@ protected RowData getBeforeImage(RowKind rowKind, HoodieCDCLogRecord cdcRecor } } + private static void closeSuppressing(ClosableIterator iterator, Throwable primary) { + try { + iterator.close(); + } catch (RuntimeException | Error closeError) { + primary.addSuppressed(closeError); + } + } + // ------------------------------------------------------------------------- // REPLACE_COMMIT // ------------------------------------------------------------------------- diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/cdc/TestCdcImageManager.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/cdc/TestCdcImageManager.java index cb7398420ccaa..beb4252c822f5 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/cdc/TestCdcImageManager.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/cdc/TestCdcImageManager.java @@ -34,6 +34,8 @@ import org.apache.flink.table.types.logical.VarCharType; import org.apache.flink.types.RowKind; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; import org.mockito.MockedStatic; import java.io.ByteArrayOutputStream; @@ -50,6 +52,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockStatic; import static org.mockito.Mockito.times; @@ -154,17 +157,27 @@ void testImageCacheReuseEvictionAndClose() throws IOException { } } - @Test - void testLoadClosesImageCacheWhenIteratorCreationFails() { + private enum LoadFailure { + ITERATOR_CREATION, + ITERATION + } + + @ParameterizedTest + @EnumSource(LoadFailure.class) + void testLoadClosesImageCacheWhenLoadFails(LoadFailure mode) { HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class); when(writeConfig.getBasePath()).thenReturn("/table"); ExternalSpillableMap imageCache = mockImageCache(); - RuntimeException failure = new RuntimeException("iterator creation failed"); + ClosableIterator iterator = mockIterator(); + RuntimeException failure = new RuntimeException("load failed"); + when(iterator.hasNext()).thenThrow(failure); CdcImageManager imageManager = new CdcImageManager( - rowType("value"), - writeConfig, + rowType("value"), writeConfig, split -> { - throw failure; + if (mode == LoadFailure.ITERATOR_CREATION) { + throw failure; + } + return iterator; }); try (MockedStatic mockedFormatUtils = mockStatic(FormatUtils.class)) { @@ -175,20 +188,28 @@ void testLoadClosesImageCacheWhenIteratorCreationFails() { assertSame(failure, assertThrows( RuntimeException.class, () -> imageManager.getOrLoadImages(1024L, fileSlice("001")))); - verify(imageCache).close(); + if (mode == LoadFailure.ITERATION) { + verify(iterator).close(); + } + verify(imageCache, times(1)).close(); + imageManager.close(); + verify(imageCache, times(1)).close(); } } @Test - void testLoadClosesIteratorAndImageCacheWhenIterationFails() { + void testLoadSuppressesImageCacheCloseError() { HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class); when(writeConfig.getBasePath()).thenReturn("/table"); ExternalSpillableMap imageCache = mockImageCache(); - ClosableIterator iterator = mock(ClosableIterator.class); - RuntimeException failure = new RuntimeException("iteration failed"); - when(iterator.hasNext()).thenThrow(failure); + RuntimeException closeFailure = new RuntimeException("close failed"); + doThrow(closeFailure).when(imageCache).close(); + RuntimeException failure = new RuntimeException("load failed"); CdcImageManager imageManager = new CdcImageManager( - rowType("value"), writeConfig, split -> iterator); + rowType("value"), writeConfig, + split -> { + throw failure; + }); try (MockedStatic mockedFormatUtils = mockStatic(FormatUtils.class)) { mockedFormatUtils.when(() -> FormatUtils.spillableMap( @@ -198,11 +219,49 @@ void testLoadClosesIteratorAndImageCacheWhenIterationFails() { assertSame(failure, assertThrows( RuntimeException.class, () -> imageManager.getOrLoadImages(1024L, fileSlice("001")))); - verify(iterator).close(); - verify(imageCache).close(); + assertEquals(1, failure.getSuppressed().length, "close failure must be suppressed, not lost"); + assertSame(closeFailure, failure.getSuppressed()[0]); } } + @Test + void testCloseContinuesAfterFailureAndClearsCache() throws IOException { + HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class); + when(writeConfig.getBasePath()).thenReturn("/table"); + ExternalSpillableMap first = mockImageCache(); + ExternalSpillableMap second = mockImageCache(); + RuntimeException firstFailure = new RuntimeException("first close failed"); + RuntimeException secondFailure = new RuntimeException("second close failed"); + doThrow(firstFailure).when(first).close(); + doThrow(secondFailure).when(second).close(); + CdcImageManager imageManager = new CdcImageManager( + rowType("value"), writeConfig, + split -> ClosableIterator.wrap(List.of().iterator())); + + try (MockedStatic mockedFormatUtils = mockStatic(FormatUtils.class)) { + mockedFormatUtils.when(() -> FormatUtils.spillableMap( + writeConfig, 1024L, CdcImageManager.class.getSimpleName())) + .thenReturn(first, second); + imageManager.getOrLoadImages(1024L, fileSlice("001")); + imageManager.getOrLoadImages(1024L, fileSlice("002")); + + assertSame(firstFailure, assertThrows(RuntimeException.class, imageManager::close)); + assertEquals(1, firstFailure.getSuppressed().length); + assertSame(secondFailure, firstFailure.getSuppressed()[0]); + verify(first, times(1)).close(); + verify(second, times(1)).close(); + + imageManager.close(); + verify(first, times(1)).close(); + verify(second, times(1)).close(); + } + } + + @SuppressWarnings("unchecked") + private static ClosableIterator mockIterator() { + return mock(ClosableIterator.class); + } + @SuppressWarnings("unchecked") private static ExternalSpillableMap mockImageCache() { ExternalSpillableMap imageCache = mock(ExternalSpillableMap.class); From 311709c962d54ce437ac45fbb8fa6f7a85977798 Mon Sep 17 00:00:00 2001 From: danny0405 Date: Tue, 4 Aug 2026 09:42:31 +0800 Subject: [PATCH 3/5] fix(flink): retain CDC images across child splits --- .../hudi/table/format/cdc/CdcIterators.java | 21 +++++------ .../table/format/cdc/TestCdcIterators.java | 36 +++++++++++++++++-- 2 files changed, 43 insertions(+), 14 deletions(-) diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java index c01d74b229f66..f7fe6109bd4b6 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java @@ -146,12 +146,14 @@ public RowData next() { @Override public void close() { - if (recordIterator != null) { - recordIterator.close(); - } - if (imageManager != null) { - imageManager.close(); - imageManager = null; + ClosableIterator iterator = recordIterator; + recordIterator = null; + CdcImageManager manager = imageManager; + imageManager = null; + try (CdcImageManager ignored = manager) { + if (iterator != null) { + iterator.close(); + } } } } @@ -347,12 +349,7 @@ public RowData next() { @Override public void close() { - // Closing the shared image manager here is required for correctness: this iterator - // destructively updates cached before-images, and LOG_FILE slices from the same instant - // share a cache key. Closing forces the next split to reload its own before-images. - try (CdcImageManager ignored = imageManager) { - logRecordIterator.close(); - } + logRecordIterator.close(); } @SuppressWarnings("unchecked") diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/cdc/TestCdcIterators.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/cdc/TestCdcIterators.java index 2144f05c7c6ef..b8b6258dcc696 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/cdc/TestCdcIterators.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/cdc/TestCdcIterators.java @@ -46,7 +46,10 @@ 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.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -117,14 +120,43 @@ void testCdcFileSplitsIteratorMovesAcrossEmptySplitsAndClosesResources() { assertTrue(iterator.hasNext()); assertSame(row, iterator.next()); + verify(firstIterator).close(); + verify(imageManager, never()).close(); assertFalse(iterator.hasNext()); + verify(secondIterator).close(); + verify(imageManager, never()).close(); iterator.close(); - verify(firstIterator).close(); - verify(secondIterator).close(); verify(imageManager).close(); } + @Test + void testCdcFileSplitsIteratorSuppressesImageManagerCloseFailure() { + HoodieCDCFileSplit split = new HoodieCDCFileSplit( + "001", HoodieCDCInferenceCase.BASE_FILE_INSERT, "first.parquet"); + ClosableIterator recordIterator = mockIterator(); + when(recordIterator.hasNext()).thenReturn(true); + RuntimeException iteratorFailure = new RuntimeException("iterator close failed"); + doThrow(iteratorFailure).when(recordIterator).close(); + CdcImageManager imageManager = mock(CdcImageManager.class); + RuntimeException managerFailure = new RuntimeException("manager close failed"); + doThrow(managerFailure).when(imageManager).close(); + CdcIterators.CdcFileSplitsIterator iterator = + new CdcIterators.CdcFileSplitsIterator( + new HoodieCDCFileSplit[] {split}, imageManager, ignored -> recordIterator); + assertTrue(iterator.hasNext()); + + assertSame(iteratorFailure, assertThrows(RuntimeException.class, iterator::close)); + assertEquals(1, iteratorFailure.getSuppressed().length); + assertSame(managerFailure, iteratorFailure.getSuppressed()[0]); + verify(recordIterator, times(1)).close(); + verify(imageManager, times(1)).close(); + + iterator.close(); + verify(recordIterator, times(1)).close(); + verify(imageManager, times(1)).close(); + } + @Test void testReplaceCommitIteratorReadsBeforeSlice() { FileSlice beforeSlice = fileSlice(); From 0f4f3a097833fa485d889e073ff8b8113d981140 Mon Sep 17 00:00:00 2001 From: danny0405 Date: Tue, 4 Aug 2026 09:51:15 +0800 Subject: [PATCH 4/5] refactor(flink): simplify CDC iterator cleanup --- .../apache/hudi/table/format/cdc/CdcIterators.java | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java index f7fe6109bd4b6..4f4fdfa17eb5d 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java @@ -146,14 +146,13 @@ public RowData next() { @Override public void close() { - ClosableIterator iterator = recordIterator; - recordIterator = null; - CdcImageManager manager = imageManager; - imageManager = null; - try (CdcImageManager ignored = manager) { - if (iterator != null) { - iterator.close(); + try (CdcImageManager ignored = imageManager) { + if (recordIterator != null) { + recordIterator.close(); } + } finally { + recordIterator = null; + imageManager = null; } } } From f96df25eaa6af0f0c3ac2c0e93d3f0fa7e99dfe6 Mon Sep 17 00:00:00 2001 From: danny0405 Date: Tue, 4 Aug 2026 10:04:08 +0800 Subject: [PATCH 5/5] refactor(common): share close suppression helper --- .../hudi/common/util/CloseableUtils.java | 35 ++++++++++++++++ .../hudi/common/util/TestCloseableUtils.java | 40 +++++++++++++++++++ .../HoodieCdcSplitReaderFunction.java | 10 +---- .../function/HoodieSplitReaderFunction.java | 11 +---- .../table/format/cdc/CdcImageManager.java | 11 +---- .../hudi/table/format/cdc/CdcInputFormat.java | 10 +---- .../hudi/table/format/cdc/CdcIterators.java | 9 +---- 7 files changed, 83 insertions(+), 43 deletions(-) create mode 100644 hudi-common/src/main/java/org/apache/hudi/common/util/CloseableUtils.java create mode 100644 hudi-common/src/test/java/org/apache/hudi/common/util/TestCloseableUtils.java diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/CloseableUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/util/CloseableUtils.java new file mode 100644 index 0000000000000..50b6e2e6ee80b --- /dev/null +++ b/hudi-common/src/main/java/org/apache/hudi/common/util/CloseableUtils.java @@ -0,0 +1,35 @@ +/* + * 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.hudi.common.util; + +/** Utility methods for closing resources. */ +public final class CloseableUtils { + + private CloseableUtils() { + } + + /** Closes {@code closeable}, attaching any failure to {@code primary} as a suppressed exception. */ + public static void closeSuppressing(AutoCloseable closeable, Throwable primary) { + try { + closeable.close(); + } catch (Throwable closeError) { + primary.addSuppressed(closeError); + } + } +} diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/TestCloseableUtils.java b/hudi-common/src/test/java/org/apache/hudi/common/util/TestCloseableUtils.java new file mode 100644 index 0000000000000..f18339047905d --- /dev/null +++ b/hudi-common/src/test/java/org/apache/hudi/common/util/TestCloseableUtils.java @@ -0,0 +1,40 @@ +/* + * 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.hudi.common.util; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; + +class TestCloseableUtils { + + @Test + void testCloseSuppressing() { + IOException primary = new IOException("primary"); + IOException closeError = new IOException("close"); + + CloseableUtils.closeSuppressing(() -> { + throw closeError; + }, primary); + + assertArrayEquals(new Throwable[] {closeError}, primary.getSuppressed()); + } +} diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieCdcSplitReaderFunction.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieCdcSplitReaderFunction.java index f865d88e51469..4c595fbacd108 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieCdcSplitReaderFunction.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieCdcSplitReaderFunction.java @@ -66,6 +66,8 @@ import java.util.function.Function; import java.util.stream.Collectors; +import static org.apache.hudi.common.util.CloseableUtils.closeSuppressing; + /** * CDC reader function for source V2. Reads CDC splits ({@link HoodieCdcSourceSplit}) and * emits change-log {@link RowData} records tagged with the appropriate {@link org.apache.flink.types.RowKind}. @@ -243,14 +245,6 @@ conf, getHadoopConf(), tablePath, tableSchema, requiredSchema, } } - private static void closeSuppressing(ClosableIterator iterator, Throwable primary) { - try { - iterator.close(); - } catch (RuntimeException | Error closeError) { - primary.addSuppressed(closeError); - } - } - /** Reads the full-schema before/after image for a file slice (emitDelete=false). */ private ClosableIterator getFileSliceIterator(MergeOnReadInputSplit split) { FileSlice fileSlice = buildFileSlice(split); diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieSplitReaderFunction.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieSplitReaderFunction.java index 0da1dc061d55e..4af55f78a4bcf 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieSplitReaderFunction.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieSplitReaderFunction.java @@ -44,6 +44,8 @@ import java.util.List; import java.util.stream.Collectors; +import static org.apache.hudi.common.util.CloseableUtils.closeSuppressing; + /** * Default reader function implementation for both MOR and COW tables. */ @@ -89,15 +91,6 @@ protected ClosableIterator createRecordIterator(HoodieSourceSplit split } } - /** Closes {@code reader}, attaching any close failure to {@code primary} as a suppressed exception. */ - private static void closeSuppressing(HoodieRecordReader reader, Throwable primary) { - try { - reader.close(); - } catch (Exception closeError) { - primary.addSuppressed(closeError); - } - } - @Override protected RowType producedRowType() { return HoodieSchemaConverter.convertToRowType(requiredSchema); diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcImageManager.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcImageManager.java index cdc72f7798f21..726c9b672abcc 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcImageManager.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcImageManager.java @@ -45,6 +45,7 @@ import java.util.TreeMap; import java.util.function.Function; +import static org.apache.hudi.common.util.CloseableUtils.closeSuppressing; import static org.apache.hudi.hadoop.utils.HoodieInputFormatUtils.HOODIE_RECORD_KEY_COL_POS; /** @@ -112,16 +113,6 @@ private ExternalSpillableMap loadImageRecords( return imageRecordsMap; } - private static void closeSuppressing( - ExternalSpillableMap spillableMap, - Throwable primary) { - try { - spillableMap.close(); - } catch (RuntimeException | Error closeError) { - primary.addSuppressed(closeError); - } - } - public RowData getImageRecord( String recordKey, Map imageCache, diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcInputFormat.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcInputFormat.java index f7c5c9cef1832..cdafe8966abde 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcInputFormat.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcInputFormat.java @@ -48,6 +48,8 @@ import java.util.List; import java.util.function.Function; +import static org.apache.hudi.common.util.CloseableUtils.closeSuppressing; + /** * The base InputFormat class to read Hoodie data set as change logs. */ @@ -176,14 +178,6 @@ private ClosableIterator getRecordIterator( } } - private static void closeSuppressing(ClosableIterator iterator, Throwable primary) { - try { - iterator.close(); - } catch (RuntimeException | Error closeError) { - primary.addSuppressed(closeError); - } - } - /** * Get a {@link HoodieRecord} iterator using a {@link HoodieRecordReader}. * diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java index 4f4fdfa17eb5d..00bd8879039a8 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java @@ -83,6 +83,7 @@ import java.util.function.Function; import java.util.stream.Collectors; +import static org.apache.hudi.common.util.CloseableUtils.closeSuppressing; import static org.apache.hudi.table.format.FormatUtils.buildAvroRecordBySchema; /** @@ -733,14 +734,6 @@ protected RowData getBeforeImage(RowKind rowKind, HoodieCDCLogRecord cdcRecor } } - private static void closeSuppressing(ClosableIterator iterator, Throwable primary) { - try { - iterator.close(); - } catch (RuntimeException | Error closeError) { - primary.addSuppressed(closeError); - } - } - // ------------------------------------------------------------------------- // REPLACE_COMMIT // -------------------------------------------------------------------------