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 cd4ea053bc7d6..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}. @@ -223,10 +225,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( 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 2854498c9bdab..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; /** @@ -105,13 +106,16 @@ 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; } public RowData getImageRecord( String recordKey, - ExternalSpillableMap imageCache, + Map imageCache, RowKind rowKind) { byte[] bytes = imageCache.get(recordKey); ValidationUtils.checkState(bytes != null, @@ -127,7 +131,7 @@ public RowData getImageRecord( public void updateImageRecord( String recordKey, - ExternalSpillableMap imageCache, + Map imageCache, RowData row) { ByteArrayOutputStream baos = new ByteArrayOutputStream(4096); try { @@ -140,7 +144,7 @@ public void updateImageRecord( public RowData removeImageRecord( String recordKey, - ExternalSpillableMap imageCache) { + Map imageCache) { byte[] bytes = imageCache.remove(recordKey); if (bytes == null) { return null; @@ -154,8 +158,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..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. */ @@ -157,11 +159,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(), 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..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 @@ -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,9 +79,11 @@ 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; +import static org.apache.hudi.common.util.CloseableUtils.closeSuppressing; import static org.apache.hudi.table.format.FormatUtils.buildAvroRecordBySchema; /** @@ -146,11 +147,12 @@ public RowData next() { @Override public void close() { - if (recordIterator != null) { - recordIterator.close(); - } - if (imageManager != null) { - imageManager.close(); + try (CdcImageManager ignored = imageManager) { + if (recordIterator != null) { + recordIterator.close(); + } + } finally { + recordIterator = null; imageManager = null; } } @@ -253,7 +255,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 +289,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(); } } @@ -348,7 +350,6 @@ public RowData next() { @Override public void close() { logRecordIterator.close(); - imageManager.close(); } @SuppressWarnings("unchecked") @@ -662,7 +663,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 { 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..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,6 +157,111 @@ void testImageCacheReuseEvictionAndClose() throws IOException { } } + 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(); + ClosableIterator iterator = mockIterator(); + RuntimeException failure = new RuntimeException("load failed"); + when(iterator.hasNext()).thenThrow(failure); + CdcImageManager imageManager = new CdcImageManager( + rowType("value"), writeConfig, + split -> { + if (mode == LoadFailure.ITERATOR_CREATION) { + throw failure; + } + return 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")))); + if (mode == LoadFailure.ITERATION) { + verify(iterator).close(); + } + verify(imageCache, times(1)).close(); + imageManager.close(); + verify(imageCache, times(1)).close(); + } + } + + @Test + void testLoadSuppressesImageCacheCloseError() { + HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class); + when(writeConfig.getBasePath()).thenReturn("/table"); + ExternalSpillableMap imageCache = mockImageCache(); + 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 -> { + 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")))); + 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); 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();