This is an automated email from the ASF dual-hosted git repository.
danny0405 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 092e4c40c504 fix(flink): reclaim empty checkpoint caches in record
index (#20241)
092e4c40c504 is described below
commit 092e4c40c504723f6b534b994ca950b6df131bc0
Author: Shuo Cheng <[email protected]>
AuthorDate: Wed Oct 14 10:53:40 2026 +0800
fix(flink): reclaim empty checkpoint caches in record index (#20241)
* fix(flink): reclaim empty checkpoint caches in record index
* refactor(flink): separate empty cache cleanup from memory eviction
---------
Co-authored-by: danny0405 <[email protected]>
---
.../sink/partitioner/index/RecordIndexCache.java | 25 ++++++--
.../partitioner/index/TestRecordIndexCache.java | 69 ++++++++++++++++++++++
2 files changed, 88 insertions(+), 6 deletions(-)
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/RecordIndexCache.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/RecordIndexCache.java
index 8f78bd70d34f..1084c7534841 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/RecordIndexCache.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/RecordIndexCache.java
@@ -36,7 +36,7 @@ import org.apache.flink.configuration.Configuration;
import java.io.Closeable;
import java.io.IOException;
import java.util.Comparator;
-import java.util.NavigableMap;
+import java.util.Map;
import java.util.TreeMap;
/**
@@ -182,16 +182,29 @@ public class RecordIndexCache implements Closeable {
}
/**
- * Performs the actual cleaning to release memory for new caches.
+ * Removes evictable empty caches and releases memory for new caches when
needed.
*
* @param nextCacheSize the size for the next new cache
*/
private void cleanIfNecessary(long nextCacheSize) {
+ // Empty caches provide no lookup benefit, regardless of memory pressure.
+ caches.entrySet().removeIf(entry -> {
+ if (entry.getKey() < minRetainedCheckpointId &&
entry.getValue().isEmpty()) {
+ entry.getValue().close();
+ log.info("Clean record index cache for checkpoint: {}",
entry.getKey());
+ return true;
+ }
+ return false;
+ });
+
+ // Evict remaining caches oldest first, only under memory pressure.
+ long usedMemory = getInMemoryMapSize();
while (!caches.isEmpty() && caches.lastKey() < minRetainedCheckpointId
- && getInMemoryMapSize() + nextCacheSize > this.maxCacheSizeInBytes) {
- NavigableMap.Entry<Long, ExternalSpillableMap<String,
HoodieRecordGlobalLocation>> lastEntry = caches.pollLastEntry();
- lastEntry.getValue().close();
- log.info("Clean record index cache for checkpoint: {}",
lastEntry.getKey());
+ && usedMemory + nextCacheSize > maxCacheSizeInBytes) {
+ Map.Entry<Long, ExternalSpillableMap<String,
HoodieRecordGlobalLocation>> entry = caches.pollLastEntry();
+ usedMemory -= entry.getValue().getCurrentInMemoryMapSize();
+ entry.getValue().close();
+ log.info("Clean record index cache for checkpoint: {}", entry.getKey());
}
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestRecordIndexCache.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestRecordIndexCache.java
index d490a9c6ab96..f15444c015ff 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestRecordIndexCache.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestRecordIndexCache.java
@@ -19,6 +19,8 @@
package org.apache.hudi.sink.partitioner.index;
import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
+import org.apache.hudi.common.serialization.DefaultSerializer;
+import org.apache.hudi.common.util.DefaultSizeEstimator;
import org.apache.hudi.common.util.collection.ExternalSpillableMap;
import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.utils.TestConfigurations;
@@ -194,6 +196,73 @@ public class TestRecordIndexCache {
assertTrue(cache.getCaches().containsKey(4L));
}
+ @Test
+ void testEmptyCheckpointsDoNotAccumulate() {
+ for (long checkpointId = 2L; checkpointId <= 1000L; checkpointId++) {
+ cache.addCheckpointCache(checkpointId);
+ cache.markAsEvictable(checkpointId);
+ assertEquals(2, cache.getCaches().size(), "Cleanup on checkpoint
creation retains the previous and current checkpoints");
+ assertTrue(cache.getCaches().containsKey(checkpointId - 1));
+ assertTrue(cache.getCaches().containsKey(checkpointId));
+ }
+ assertNull(cache.get("missing"));
+ HoodieRecordGlobalLocation location = new
HoodieRecordGlobalLocation("partition1", "1001", "file_id1");
+ cache.update("key", location);
+ assertEquals(location, cache.get("key"));
+ }
+
+ @Test
+ void testEmptyCheckpointCleanupPreservesNonEmptyAndInflightCaches() {
+ HoodieRecordGlobalLocation oldLocation = new
HoodieRecordGlobalLocation("partition1", "1001", "file_id1");
+ HoodieRecordGlobalLocation newLocation = new
HoodieRecordGlobalLocation("partition1", "1002", "file_id2");
+ cache.update("key", oldLocation);
+ cache.update("old_key", oldLocation);
+ cache.addCheckpointCache(2L);
+ ExternalSpillableMap<String, HoodieRecordGlobalLocation> emptyCache =
Mockito.spy(cache.getCaches().get(2L));
+ cache.getCaches().put(2L, emptyCache);
+ cache.addCheckpointCache(3L);
+ cache.update("key", newLocation);
+ cache.addCheckpointCache(4L);
+
+ // An older non-empty cache must not prevent cleanup of newer evictable
empty caches.
+ cache.markAsEvictable(3L);
+ cache.addCheckpointCache(5L);
+ assertEquals(4, cache.getCaches().size());
+ assertFalse(cache.getCaches().containsKey(2L));
+ Mockito.verify(emptyCache).close();
+ assertTrue(cache.getCaches().containsKey(4L), "Empty inflight checkpoints
must be retained");
+ assertTrue(cache.getCaches().containsKey(5L));
+ assertEquals(oldLocation, cache.get("old_key"));
+ assertEquals(newLocation, cache.get("key"));
+
+ cache.markAsEvictable(5L);
+ cache.addCheckpointCache(6L);
+ assertEquals(4, cache.getCaches().size());
+ assertFalse(cache.getCaches().containsKey(4L));
+ assertTrue(cache.getCaches().containsKey(5L), "The checkpoint at the
retention boundary must remain");
+ assertTrue(cache.getCaches().containsKey(6L));
+ assertEquals(oldLocation, cache.get("old_key"));
+ assertEquals(newLocation, cache.get("key"));
+ }
+
+ @Test
+ void testEmptyCheckpointCleanupPreservesDiskOnlyCache() throws IOException {
+ ExternalSpillableMap<String, HoodieRecordGlobalLocation> diskOnlyCache =
new ExternalSpillableMap<>(
+ 0L, tempDir.getAbsolutePath(), new DefaultSizeEstimator<>(), new
DefaultSizeEstimator<>(),
+ ExternalSpillableMap.DiskMapType.ROCKS_DB, new DefaultSerializer<>(),
false, "disk-only-cache");
+ cache.getCaches().put(1L, diskOnlyCache).close();
+ HoodieRecordGlobalLocation location = new
HoodieRecordGlobalLocation("partition1", "1001", "file_id1");
+ cache.update("key", location);
+ assertEquals(0L, diskOnlyCache.getCurrentInMemoryMapSize());
+ assertEquals(1, diskOnlyCache.getDiskBasedMapNumEntries());
+
+ cache.addCheckpointCache(2L);
+ cache.markAsEvictable(2L);
+ cache.addCheckpointCache(3L);
+ assertTrue(cache.getCaches().containsKey(1L), "A cache with spilled
records is not empty");
+ assertEquals(location, cache.get("key"));
+ }
+
@Test
void testClose() throws IOException {
cache.addCheckpointCache(2L);
