acrites opened a new pull request, #40716: URL: https://github.com/apache/beam/pull/40716
* Prevented Silent Data Loss on Partial Mutations: Added the missing `return;` after `cache.remove(IterableCacheKey.INSTANCE);` in both `remove()` and `appendHelper()`. When only a partial prefix is cached (`nextToken != null`), mutating operations now properly evict the partial cache entry without re-caching truncated blocks. * Replaced `UncachedStateIterable.DecodingIterator` with `DataStreamDecoder`: Replaced the custom `DecodingIterator` with `DataStreams.DataStreamDecoder`, enabling continuous stream decoding across `ByteString` chunk boundaries and supporting multi-chunk elements. * Fixed `PrefetchableIterator.isReady()` Contract in `CachingStateIterator`: Delegated to `underlyingStateFetchingIterator.isReady()` when a block is not present in cache (instead of unconditionally returning false), allowing async prefetching from the State API to transition `isReady()` to true. * Graceful Handling of Concurrently Mutated Cache in `hasNext()`: Replaced `checkState(existing instanceof BlocksPrefix)` with a fallback branch that safely loads the next block from the State API when the cached entry was replaced by `MutatedBlocks` or `EmptyBlocks`. * Cancelled In-Flight RPCs on Seek: In `seekToContinuationToken()`, `prefetchedResponse.cancel(true)` is now called before clearing the reference to prevent abandoned RPC responses from leaking bandwidth and buffers. * Exact Capacity Allocation in `hasNext()`: Changed `new ArrayList<>(currentBlockIndex + 1)` to `new ArrayList<>(existing.getBlocks().size() + 1)` to prevent internal `ArrayList` reallocation. The newly added unit tests all demonstrate the mentioned bugs. -- 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]
