[Bug] CachingSeekableInputStream leaks a remote stream under concurrent vectored reads

[wombatu-kun]

Search before asking

• I searched in the issues and found nothing similar.

Paimon version

master, 142f823

Compute Engine

Engine independent. The path is in paimon-common, so it is reachable from Flink, Spark and the Java API alike.

Minimal reproduce step

Found by code inspection rather than from a failing job. Read a file through the caching file IO (CachingFileIO, block cache enabled) using vectored reads that span at least two ranges, for example any Parquet row group with several column chunks.

What doesn't meet your expectations?

CachingSeekableInputStream.getRemoteStream() lazily creates the remote stream into a plain, non-volatile field with no locking:

@Nullable private SeekableInputStream remoteStream;                 // line 39

private SeekableInputStream getRemoteStream() throws IOException {  // line 180
    if (remoteStream == null) {
        remoteStream = fileIO.newInputStream(path);
    }
    return remoteStream;
}

That method is reached concurrently. CachingSeekableInputStream implements VectoredReadable, and VectoredReadUtils.readVectored submits one readSingleRange task per range onto IO_THREAD_POOL (VectoredReadUtils lines 84 and 88). Each task calls back into pread / preadFully on the same instance, which goes readBlock -> readRemote -> getRemoteStream().

Two problems follow:

1. Leaked stream. Two threads can both observe null, both call fileIO.newInputStream(path), and one assignment overwrites the other. The losing stream is then unreachable and never closed, because close() only closes whatever the field happens to hold.
2. Unsafe publication. The field is not volatile, so a thread can observe a reference to a stream whose construction is not yet visible to it.

Anything else?

This interacts with #8962, which gives the RESTTokenFileIO cache real lifetime tracking. A stream that is never closed holds its lease forever, so that entry's FileIO is never released. That is the same outcome as today (the eviction listener is currently inert, which is what #8962 fixes), so this is not a regression - it means the fix simply does not reach the affected entries.

Fix shape: guard the lazy initialisation, either with double-checked locking on a volatile field or by creating the stream eagerly, closing the loser if two are ever built.

Are you willing to submit a PR?

• I'm willing to submit a PR!