(beam) branch master updated: [IcebergIO] Fix stale openWriters count in RecordWriterManager (#39157)

[ahmedabualsaud]

This is an automated email from the ASF dual-hosted git repository.

ahmedabu98 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new 961ed7c31ad [IcebergIO] Fix stale openWriters count in 
RecordWriterManager (#39157)
961ed7c31ad is described below

commit 961ed7c31adec4dec63fca5687b4b0083107da46
Author: atognolas <togno...@spotify.com>
AuthorDate: Mon Jun 29 22:07:27 2026 +0200

    [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) <nore...@anthropic.com>
---
 .../java/org/apache/beam/sdk/io/iceberg/RecordWriterManager.java     | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git 
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriterManager.java
 
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriterManager.java
index 1e25e6b9c23..d5439ad8383 100644
--- 
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriterManager.java
+++ 
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriterManager.java
@@ -163,7 +163,10 @@ class RecordWriterManager implements AutoCloseable {
 
       @Nullable RecordWriter writer = 
writers.getIfPresent(routingPartitionKey);
       if (writer == null && openWriters >= maxNumWriters) {
-        return false;
+        writers.cleanUp();
+        if (openWriters >= maxNumWriters) {
+          return false;
+        }
       }
       writer = fetchWriterForPartition(routingPartitionKey, writer);
       writer.write(record);