danny0405 commented on code in PR #19482:
URL: https://github.com/apache/hudi/pull/19482#discussion_r3708690040
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcImageManager.java:
##########
@@ -105,13 +110,26 @@ private ExternalSpillableMap<String, byte[]>
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(
Review Comment:
Addressed in adb73870b52e: CdcImageManager.close now attempts every
cached-map close, preserves the first failure with later failures suppressed,
and clears the cache in finally. Added coverage for continued cleanup and cache
clearing.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcImageManager.java:
##########
@@ -105,13 +110,26 @@ private ExternalSpillableMap<String, byte[]>
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<String, byte[]> imageRecordsMap,
Review Comment:
Addressed in adb73870b52e: renamed the helper parameter to spillableMap so
the signature is self-contained.
##########
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<String, byte[]> imageCache = mockImageCache();
+ RuntimeException failure = new RuntimeException("iterator creation
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"))));
+ verify(imageCache).close();
+ }
+ }
+
+ @Test
+ void testLoadClosesIteratorAndImageCacheWhenIterationFails() {
+ HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class);
+ when(writeConfig.getBasePath()).thenReturn("/table");
+ ExternalSpillableMap<String, byte[]> imageCache = mockImageCache();
+ ClosableIterator<RowData> iterator = mock(ClosableIterator.class);
Review Comment:
Addressed in adb73870b52e: extracted mockIterator with
SuppressWarnings("unchecked") and use it in the parameterized test.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcImageManager.java:
##########
@@ -105,13 +110,26 @@ private ExternalSpillableMap<String, byte[]>
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<String, byte[]> imageRecordsMap,
+ Throwable primary) {
+ try {
+ imageRecordsMap.close();
+ } catch (RuntimeException | Error closeError) {
+ primary.addSuppressed(closeError);
Review Comment:
Addressed in adb73870b52e: added testLoadSuppressesImageCacheCloseError,
which verifies that the original load failure is rethrown and the close failure
is attached as its suppressed exception.
##########
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<String, byte[]> imageCache = mockImageCache();
+ RuntimeException failure = new RuntimeException("iterator creation
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"))));
+ verify(imageCache).close();
+ }
+ }
+
+ @Test
+ void testLoadClosesIteratorAndImageCacheWhenIterationFails() {
Review Comment:
Addressed in adb73870b52e: collapsed the duplicate failure tests into one
parameterized test over iterator-creation and iteration failures.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java:
##########
@@ -347,8 +347,9 @@ public RowData next() {
@Override
public void close() {
- logRecordIterator.close();
- imageManager.close();
+ try (CdcImageManager ignored = imageManager) {
+ logRecordIterator.close();
+ }
}
Review Comment:
Addressed in adb73870b52e: added a comment explaining that the per-split
close is required because the iterator mutates cached before-images and
same-instant LOG_FILE slices share the cache key, so the next split must reload.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcImageManager.java:
##########
@@ -86,7 +86,12 @@ public ExternalSpillableMap<String, byte[]> getOrLoadImages(
cache.remove(oldest).close();
}
ExternalSpillableMap<String, byte[]> images =
loadImageRecords(maxCompactionMemoryInBytes, fileSlice);
- cache.put(instant, images);
+ try {
+ cache.put(instant, images);
+ } catch (RuntimeException | Error e) {
+ closeSuppressing(images, e);
+ throw e;
+ }
Review Comment:
Addressed in adb73870b52e: removed the unreachable TreeMap.put try/catch.
##########
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<String, byte[]> imageCache = mockImageCache();
+ RuntimeException failure = new RuntimeException("iterator creation
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"))));
+ verify(imageCache).close();
Review Comment:
Addressed in adb73870b52e: the parameterized test now uses times(1), calls
imageManager.close, and verifies the failed map is not closed again, proving it
was not cached.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java:
##########
@@ -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);
Review Comment:
Addressed in adb73870b52e: both LOG_FILE construction call sites now close
recordIterator on constructor or argument-evaluation failure while suppressing
cleanup errors. BeforeImageIterator also closes its already-open cdcItr if
image initialization fails.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
