This is an automated email from the ASF dual-hosted git repository.
Zakelly pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 89265005f9d [FLINK-40645][state/forst] Unregister closed streams from
FileCacheEntry#openedStreams (#29169)
89265005f9d is described below
commit 89265005f9d834e9596f54efdb82ab9162cd49c0
Author: Seungjoo Choi <[email protected]>
AuthorDate: Wed Oct 7 21:50:35 2026 +0900
[FLINK-40645][state/forst] Unregister closed streams from
FileCacheEntry#openedStreams (#29169)
Generated-by: Claude Code (Claude Fable 5.1)
---
.../forst/fs/cache/CachedDataInputStream.java | 5 +
.../flink/state/forst/fs/cache/FileCacheEntry.java | 13 ++
.../state/forst/fs/cache/FileCacheEntryTest.java | 162 +++++++++++++++++++++
3 files changed, 180 insertions(+)
diff --git
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/CachedDataInputStream.java
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/CachedDataInputStream.java
index da070128f0e..c9bd59cc262 100644
---
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/CachedDataInputStream.java
+++
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/CachedDataInputStream.java
@@ -306,6 +306,11 @@ public class CachedDataInputStream extends
FSDataInputStream implements ByteBuff
return;
}
closed = true;
+ // FileCacheEntry#doRemoveFile only drops closed streams when the file
itself is evicted
+ // or deleted, so while a file stays cached every open/close cycle of
a stream on it would
+ // otherwise leave a closed stream (and the remote stream it wraps) on
the heap. Done
+ // before the closes below so that a failing close cannot skip it.
+ cacheEntry.unregisterStream(this);
// Both the cached stream and the original (remote) stream belong to
this wrapper:
// FileCacheEntry#open hands the original stream over and keeps no
reference to it, so
// nobody else can close it. Leaving it open leaks whatever the remote
file system holds
diff --git
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/FileCacheEntry.java
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/FileCacheEntry.java
index 4b1e9c70672..00e56d273a2 100644
---
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/FileCacheEntry.java
+++
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/FileCacheEntry.java
@@ -147,6 +147,19 @@ public final class FileCacheEntry extends
ReferenceCounted<Object> {
}
}
+ /**
+ * Removes a closed stream from the queue of opened streams, so that it
does not stay there
+ * until the file is evicted or deleted.
+ *
+ * @param stream the closed stream to unregister
+ */
+ void unregisterStream(CachedDataInputStream stream) {
+ // Deliberately not synchronized: LinkedBlockingQueue#remove is safe
against the concurrent
+ // iteration in doRemoveFile(), while taking the entry lock would
serialize every stream
+ // close against cache eviction.
+ openedStreams.remove(stream);
+ }
+
/**
* Retrieves the cached input stream for this cache entry if it is
available and the entry is in
* a valid state. The method attempts to open the cached stream if the
entry is in the LOADED
diff --git
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/cache/FileCacheEntryTest.java
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/cache/FileCacheEntryTest.java
new file mode 100644
index 00000000000..c4baa8b3b8e
--- /dev/null
+++
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/cache/FileCacheEntryTest.java
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.forst.fs.cache;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for the bookkeeping of opened streams in {@link FileCacheEntry}. */
+class FileCacheEntryTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ /**
+ * A closed stream must leave {@link FileCacheEntry#openedStreams}
immediately. Before, closed
+ * streams were only dropped by {@link FileCacheEntry#doRemoveFile()},
i.e. when the file was
+ * evicted or deleted, so they accumulated on the heap for as long as the
file stayed cached.
+ */
+ @Test
+ void testClosedStreamIsUnregisteredFromEntry() throws Exception {
+ FileBasedCache cache =
+ new FileBasedCache(
+ new Configuration(),
+ new SizeBasedCacheLimitPolicy(1024 * 1024, 64 * 1024 *
1024),
+ FileSystem.getLocalFileSystem(),
+ new Path(tempDir.toString(), "cache"),
+ null);
+ try {
+ FileCacheEntry entry =
+ new FileCacheEntry(
+ cache,
+ new Path(tempDir.toString(), "remote/1.sst"),
+ new Path(tempDir.toString(), "cache/1.sst"),
+ 16);
+
+ CachedDataInputStream first = entry.open(new NoopStream());
+ CachedDataInputStream second = entry.open(new NoopStream());
+ assertThat(entry.openedStreams).containsExactly(first, second);
+
+ first.close();
+ assertThat(entry.openedStreams)
+ .as("closed stream must not stay registered")
+ .containsExactly(second);
+
+ // Closing again must not fail or touch the registration of the
other stream.
+ first.close();
+ assertThat(entry.openedStreams).containsExactly(second);
+
+ second.close();
+ assertThat(entry.openedStreams).isEmpty();
+ } finally {
+ cache.close();
+ }
+ }
+
+ /**
+ * The wrapper is unregistered before the cached stream is closed, so a
failure while closing
+ * the cached stream must not leave the (now unusable) wrapper behind in
{@link
+ * FileCacheEntry#openedStreams}. A real cached stream from {@link
FileCacheEntry#open} never
+ * fails to close, so the CACHED_OPEN stream is built directly with a
stream that throws.
+ */
+ @Test
+ void testClosedStreamIsUnregisteredWhenCachedCloseThrows() throws
Exception {
+ FileBasedCache cache =
+ new FileBasedCache(
+ new Configuration(),
+ new SizeBasedCacheLimitPolicy(1024 * 1024, 64 * 1024 *
1024),
+ FileSystem.getLocalFileSystem(),
+ new Path(tempDir.toString(), "cache"),
+ null);
+ try {
+ FileCacheEntry entry =
+ new FileCacheEntry(
+ cache,
+ new Path(tempDir.toString(), "remote/1.sst"),
+ new Path(tempDir.toString(), "cache/1.sst"),
+ 16);
+
+ // CACHED_OPEN stream (constructor with a cache stream); its
cached stream's close()
+ // throws, so close() propagates from closeCachedStream() ->
fsdis.close().
+ CachedDataInputStream wrapper =
+ new CachedDataInputStream(
+ cache, entry, new ThrowingOnCloseStream(), new
NoopStream());
+ entry.openedStreams.add(wrapper);
+ assertThat(entry.openedStreams).containsExactly(wrapper);
+
+ assertThatThrownBy(wrapper::close).isInstanceOf(IOException.class);
+
+ assertThat(entry.openedStreams)
+ .as("a failing cached-stream close must still unregister
the wrapper")
+ .doesNotContain(wrapper);
+ } finally {
+ cache.close();
+ }
+ }
+
+ private static final class NoopStream extends FSDataInputStream {
+ private long pos;
+
+ @Override
+ public void seek(long desired) {
+ pos = desired;
+ }
+
+ @Override
+ public long getPos() {
+ return pos;
+ }
+
+ @Override
+ public int read() {
+ pos++;
+ return 0;
+ }
+ }
+
+ /** A stream whose {@link #close()} always fails, to exercise the
failing-close path. */
+ private static final class ThrowingOnCloseStream extends FSDataInputStream
{
+ @Override
+ public void seek(long desired) {}
+
+ @Override
+ public long getPos() {
+ return 0;
+ }
+
+ @Override
+ public int read() {
+ return 0;
+ }
+
+ @Override
+ public void close() throws IOException {
+ throw new IOException("close failed");
+ }
+ }
+}
