This is an automated email from the ASF dual-hosted git repository.
voonhous 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 f41e8e3070d9 fix(flink): close CDC image spillable maps on failures
(#19482)
f41e8e3070d9 is described below
commit f41e8e3070d94c7997eed933ab76ceb250d77e53
Author: Danny Chan <[email protected]>
AuthorDate: Tue Aug 4 22:51:16 2026 +0800
fix(flink): close CDC image spillable maps on failures (#19482)
* fix(flink): close CDC image spillable maps on failures
* fix(flink): harden CDC cleanup failure paths
* fix(flink): retain CDC images across child splits
* refactor(flink): simplify CDC iterator cleanup
* refactor(common): share close suppression helper
---
.../apache/hudi/common/util/CloseableUtils.java | 35 +++++++
.../hudi/common/util/TestCloseableUtils.java | 40 ++++++++
.../function/HoodieCdcSplitReaderFunction.java | 15 ++-
.../reader/function/HoodieSplitReaderFunction.java | 11 +--
.../hudi/table/format/cdc/CdcImageManager.java | 31 +++++-
.../hudi/table/format/cdc/CdcInputFormat.java | 17 +++-
.../apache/hudi/table/format/cdc/CdcIterators.java | 30 +++---
.../hudi/table/format/cdc/TestCdcImageManager.java | 108 +++++++++++++++++++++
.../hudi/table/format/cdc/TestCdcIterators.java | 36 ++++++-
9 files changed, 286 insertions(+), 37 deletions(-)
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 000000000000..50b6e2e6ee80
--- /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 000000000000..f18339047905
--- /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 cd4ea053bc7d..4c595fbacd10 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.List;
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 @@ public class HoodieCdcSplitReaderFunction extends
AbstractSplitReaderFunction {
String logFilePath = new Path(tablePath,
fileSplit.getCdcFiles().get(0)).toString();
MergeOnReadInputSplit split =
CdcIterators.singleLogFile2Split(tablePath, logFilePath,
maxCompactionMemoryInBytes);
ClosableIterator<HoodieRecord<RowData>> 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 0da1dc061d55..4af55f78a4bc 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.Collections;
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 @@ public class HoodieSplitReaderFunction extends
AbstractSplitReaderFunction {
}
}
- /** Closes {@code reader}, attaching any close failure to {@code primary} as
a suppressed exception. */
- private static void closeSuppressing(HoodieRecordReader<RowData> 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 2854498c9bda..726c9b672abc 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.Map;
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 @@ public class CdcImageManager implements AutoCloseable {
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<String, byte[]> imageCache,
+ Map<String, byte[]> imageCache,
RowKind rowKind) {
byte[] bytes = imageCache.get(recordKey);
ValidationUtils.checkState(bytes != null,
@@ -127,7 +131,7 @@ public class CdcImageManager implements AutoCloseable {
public void updateImageRecord(
String recordKey,
- ExternalSpillableMap<String, byte[]> imageCache,
+ Map<String, byte[]> imageCache,
RowData row) {
ByteArrayOutputStream baos = new ByteArrayOutputStream(4096);
try {
@@ -140,7 +144,7 @@ public class CdcImageManager implements AutoCloseable {
public RowData removeImageRecord(
String recordKey,
- ExternalSpillableMap<String, byte[]> imageCache) {
+ Map<String, byte[]> imageCache) {
byte[] bytes = imageCache.remove(recordKey);
if (bytes == null) {
return null;
@@ -154,8 +158,25 @@ public class CdcImageManager implements AutoCloseable {
@Override
public void close() {
- cache.values().forEach(ExternalSpillableMap::close);
- cache.clear();
+ RuntimeException failure = null;
+ try {
+ for (ExternalSpillableMap<String, byte[]> 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 ff810bb4906c..cdafe8966abd 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.io.IOException;
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 @@ public class CdcInputFormat extends
MergeOnReadInputFormat {
String logFilepath = new Path(tablePath,
fileSplit.getCdcFiles().get(0)).toString();
MergeOnReadInputSplit split =
CdcIterators.singleLogFile2Split(tablePath, logFilepath,
maxCompactionMemoryInBytes);
ClosableIterator<HoodieRecord<RowData>> 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 725873329bcc..00bd8879039a 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.hadoop.fs.HadoopFSUtils;
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.Arrays;
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 final class CdcIterators {
@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 final class CdcIterators {
private final String[] orderingFields;
private final TypedProperties props;
- private ExternalSpillableMap<String, byte[]> beforeImages;
+ private Map<String, byte[]> beforeImages;
private RowData currentImage;
private RowData sideImage;
@@ -287,15 +289,15 @@ public final class CdcIterators {
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 final class CdcIterators {
@Override
public void close() {
logRecordIterator.close();
- imageManager.close();
}
@SuppressWarnings("unchecked")
@@ -662,7 +663,12 @@ public final class CdcIterators {
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 e761a2a7945f..beb4252c822f 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.RowType;
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.assertSame;
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 @@ class TestCdcImageManager {
}
}
+ 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<String, byte[]> imageCache = mockImageCache();
+ ClosableIterator<RowData> 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<FormatUtils> 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<String, byte[]> 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<FormatUtils> 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<String, byte[]> first = mockImageCache();
+ ExternalSpillableMap<String, byte[]> 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.<RowData>of().iterator()));
+
+ try (MockedStatic<FormatUtils> 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<RowData> mockIterator() {
+ return mock(ClosableIterator.class);
+ }
+
@SuppressWarnings("unchecked")
private static ExternalSpillableMap<String, byte[]> mockImageCache() {
ExternalSpillableMap<String, byte[]> 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 2144f05c7c6e..b8b6258dcc69 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.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.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 @@ class TestCdcIterators {
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<RowData> 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();
