Skip to content

Commit 961ed7c

Browse files
atognolasclaude
andauthored
[IcebergIO] Fix stale openWriters count in RecordWriterManager (#39157)
RecordWriterManager uses a Guava Cache with expireAfterAccess(1 min) for writer eviction. However, Guava Cache eviction is lazy — expired entries are only removed on subsequent access or explicit cleanUp(). The write() method checks `openWriters >= maxNumWriters` without first calling cleanUp(), so expired writers remain in the cache and openWriters stays stale. With DEFAULT_MAX_WRITERS_PER_BUNDLE = 20 and more than 20 partitions, this causes 100% false spill — every record for the 21st+ partition is rejected as if the writer pool is full. Add writers.cleanUp() before the capacity check to evict expired entries and update openWriters accurately. Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 parent 402b41c commit 961ed7c

1 file changed

Lines changed: 4 additions & 1 deletion

File tree

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriterManager.java

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -163,7 +163,10 @@ boolean write(Record record) {
163163

164164
@Nullable RecordWriter writer = writers.getIfPresent(routingPartitionKey);
165165
if (writer == null && openWriters >= maxNumWriters) {
166-
return false;
166+
writers.cleanUp();
167+
if (openWriters >= maxNumWriters) {
168+
return false;
169+
}
167170
}
168171
writer = fetchWriterForPartition(routingPartitionKey, writer);
169172
writer.write(record);

0 commit comments

Comments
 (0)