voonhous commented on code in PR #19482:
URL: https://github.com/apache/hudi/pull/19482#discussion_r3704082476
##########
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:
Same leak class this PR is fixing, one frame up and still open: this
constructor can throw with a live reader already allocated by the caller.
`initImages` -> `getOrLoadImages` is declared `throws IOException`, and
`BufferedRecordMergerFactory.create` at `:279` can throw as well. Both call
sites build the log reader **before** invoking this constructor:
- `CdcInputFormat.java:159-164` -- `ClosableIterator<HoodieRecord<RowData>>
recordIterator = getSplitRecordIterator(split);` then `return new
CdcIterators.DataLogFileIterator(..., recordIterator, ...)`
- `HoodieCdcSplitReaderFunction.java:225-229` -- identical shape
If the constructor throws, `recordIterator` (an open log/base-file reader)
is never closed. `DataLogFileIterator.close()` can never run on a
half-constructed object, so nothing downstream can recover it.
`BeforeImageIterator` has the same hole: `super(...)` at `:662` opens
`cdcItr` (`:431`), then `initImages(fileSlice)` at `:666` can throw
`IOException` from `getOrLoadImages` or `HoodieException` from the
`ValidationUtils.checkState` calls at `:670-671` / `:717-718`, leaking `cdcItr`.
Action: wrap both `new DataLogFileIterator(...)` call sites in a try/catch
that closes `recordIterator` and rethrows, and wrap the `initImages(fileSlice)`
call in `BeforeImageIterator`'s constructor in a try/catch that closes `cdcItr`
and rethrows. Fine as a follow-up PR if you would rather keep this one scoped
-- but since this PR is specifically about close-on-failure in these classes,
it should not be left silent.
##########
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:
The PR body says this change is "preserving cleanup failures as suppressed
exceptions", but this line never executes in any test. `mockImageCache()`
(`TestCdcImageManager.java:206-217`) stubs only `get`/`put`/`remove`; `close()`
on a Mockito mock is a silent no-op, so both new tests take the happy path
through `closeSuppressing`. Delete the try/catch in this helper and the suite
still passes -- the advertised behavior is unpinned.
There is an in-repo template for exactly this, added by you in #19202
(`d10b868d90c9`):
`TestHoodieSplitReaderFunction#testCreateRecordIteratorSuppressesCloseError`
(`TestHoodieSplitReaderFunction.java:498-521`) covers the identical helper in
`HoodieSplitReaderFunction.java:93-99`.
Please add the mirror test. Verified to pass on this branch, and to fail
with `expected: <RuntimeException: load failed> but was: <RuntimeException:
close failed>` once the try/catch here is stripped:
```java
@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]);
}
}
```
While you are here: neither test drives the `IOException` arm of the
multi-catch at `:113` either (serializer failure) -- one `doThrow(new
IOException(...))` on the serialize path would close that out.
##########
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:
This try/catch is unreachable. `cache` is a `TreeMap<String, ...>` (`:64`,
`:73`) assigned once and never reassigned, and line 80 already calls
`cache.containsKey(instant)` on the same key -- a null key NPEs there first
(`new TreeMap<String,Object>().containsKey(null)` throws on an empty map), and
`String` is `Comparable`, so no `ClassCastException`. `TreeMap.put` has nothing
left to throw.
Neither new test reaches these lines either; both land in the
`loadImageRecords` catch at `:113`. So this is five lines of dead code that
cannot be covered.
Action: drop it.
```suggestion
cache.put(instant, images);
```
##########
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:
`close()` at `:173-177` has the exact bug this helper exists to fix:
```java
cache.values().forEach(ExternalSpillableMap::close);
cache.clear();
```
If one map's `close()` throws, every remaining map leaks **and**
`cache.clear()` never runs. Not hypothetical here: `RocksDbDiskMap.close()` ->
`rocksDb.close()` propagates (`RocksDbDiskMap.java:167-174`), and
`ExternalSpillableMap.close()` dereferences `diskBasedMap.size()` inside the
log call at `:274`. Your own new test
`testLoadClosesIteratorAndImageCacheWhenIterationFails` treats a throwing
`close()` as a real possibility, so the two methods should not disagree about
it.
Action: suppress per-entry failures and clear in a `finally`:
```java
@Override
public void close() {
RuntimeException primary = null;
try {
for (ExternalSpillableMap<String, byte[]> map : cache.values()) {
try {
map.close();
} catch (RuntimeException e) {
if (primary == null) {
primary = e;
} else {
primary.addSuppressed(e);
}
}
}
} finally {
cache.clear();
}
if (primary != null) {
throw primary;
}
}
```
##########
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:
This `close()` is load-bearing for correctness and nothing says so. Worth a
comment before this lands, because the next person to "clean it up" ships a
wrong-results bug.
`DataLogFileIterator` is the only per-split iterator that closes the shared
`CdcImageManager`. Every sibling leaves it alone (`AddBaseFileIterator` :191,
`RemoveBaseFileIterator` :228, `BaseImageIterator` :566,
`ReplaceCommitIterator` :772), and `HoodieCdcSplitReaderFunction.java:132-134`
states outright that "The CdcFileSplitsIterator owns the imageManager and its
per-split record iterators". So this reads like an ownership violation you
could just delete.
It is not -- it is masking two real bugs:
1. **Destructive mutation.** `hasNext()` drains the shared cached map via
`removeImageRecord` (:311) and writes back into it via `updateImageRecord`
(:333).
2. **Cache-key collision.** `CdcImageManager` keys on
`fileSlice.getBaseInstantTime()` (`CdcImageManager.java:79`), but
`HoodieCDCExtractor` builds every LOG_FILE before-slice as `new FileSlice(fgId,
instant.requestedTime(), baseFile, logFiles)` (`HoodieCDCExtractor.java:367`),
while `HoodieCommitMetadata.getDependentFileSliceForFileGroupFromDeltaCommit`
(`:266-277`) returns a **different log-file list per log file** for exactly the
multi-write case it documents there -- "log file rolls over" and "eager flush
from flink memory buffer". So a delta commit that flushes 2+ log files into one
file group (routine on Flink once a base file exists) yields 2+ LOG_FILE splits
whose before-slices share one cache key but describe different data.
`HoodieCDCFileSplit.compareTo:113-115` confirms these same-instant splits are
an expected, ordered case.
Both are invisible today only because this line nukes the manager between
splits: `CdcFileSplitsIterator.hasNext()` closes the exhausted child at
`CdcIterators.java:132` before building the next one, so the next split always
reloads from disk. This PR makes that masking *more* reliable (the manager now
closes even when `logRecordIterator.close()` throws) without recording why.
Please record it:
```suggestion
@Override
public void close() {
// Closing the shared image manager here is required for correctness,
not tidiness:
// hasNext() destructively drains and rewrites the cached before-image
map, and LOG_FILE
// before-slices from the same instant collide on the cache key
(FileSlice#getBaseInstantTime),
// so a stale entry would be served to the next split. Closing forces
a reload.
// Do not hand ownership back to CdcFileSplitsIterator without first
giving this iterator
// its own private before-image map.
try (CdcImageManager ignored = imageManager) {
logRecordIterator.close();
}
}
```
##########
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:
nit, feel free to ignore: no explicit invocation count, and nothing checks
that the failed load left the closed map out of the cache (a closed-but-cached
map would be handed to the next split). The surrounding convention already uses
explicit counts -- same file `:153` `verify(second, times(1)).close()`, and
`TestHoodieSplitReaderFunction.java:495` `verify(reader, times(1)).close()`.
```suggestion
verify(imageCache, times(1)).close();
// a failed load must not leave the closed map in the cache
imageManager.close();
verify(imageCache, times(1)).close();
```
Same applies at `:202`.
##########
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:
nit, feel free to ignore: unchecked conversion with no suppression. Both
this file (`:206`, `@SuppressWarnings("unchecked")` on `mockImageCache()`) and
`TestHoodieSplitReaderFunction.java:526-527` annotate this pattern.
Action: extract a helper next to `mockImageCache()` and call it here.
```java
@SuppressWarnings("unchecked")
private static ClosableIterator<RowData> mockIterator() {
return mock(ClosableIterator.class);
}
```
##########
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:
These two tests are near-duplicates -- roughly 15 identical setup lines --
and they pin the same single `catch` clause (`CdcImageManager.java:113-116`).
`verify(iterator).close()` at `:201` is also vacuous. A/B against `HEAD~1`
with this test file copied in:
```
Tests run: 5, Failures: 2
testLoadClosesImageCacheWhenIteratorCreationFails:178 Wanted but not
invoked: close()
testLoadClosesIteratorAndImageCacheWhenIterationFails:202 Wanted but not
invoked: close()
```
Pre-PR it fails at `:202`, not `:201` -- the try-with-resources in
`loadImageRecords` already closed the iterator, so that assertion documents
pre-existing behavior rather than the fix, which is why the second test reads
as new coverage when only one line of it is.
Action: collapse the two into one parameterized test. Verified to compile
and pass on this branch:
```java
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();
}
}
```
--
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]
