[DRAFT][FLIP-597][filesystem] Add common object storage stream interfaces

flink-core/src/main/java/org/apache/flink/core/fs/Abortable.java

@@ -0,0 +1,46 @@
+/*
+ * 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.core.fs;
+
+import org.apache.flink.annotation.Experimental;
+
+/**
+ * Optional capability for streams that support immediate resource release from an external thread.
+ *
+ * <p>Implemented by SDK streams that can unblock a concurrent {@code read()} or {@code write()}.
+ * Discovered via {@code instanceof}.
+ *
+ * <p>Contract:
+ *
+ * <ul>
+ *   <li>May be called concurrently with an in-progress {@code read()} or {@code write()}.
+ *   <li>Must never throw.
+ *   <li>Must be idempotent.
+ *   <li>Does not replace {@code close()}.
+ * </ul>
+ */
+@Experimental
+public interface Abortable {
+
+    /**
+     * Releases the stream's resources immediately, unblocking any blocked {@code read()} or {@code
+     * write()}.
+     */
+    void abort();
+}

flink-core/src/main/java/org/apache/flink/core/fs/BufferingInputStreamExtension.java

@@ -0,0 +1,53 @@
+/*
+ * 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.core.fs;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.concurrent.Immutable;
+
+import java.io.BufferedInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * Default {@link InputStreamExtension} that opens the raw stream via an {@link InputStreamOpener}
+ * and wraps it with a {@link BufferedInputStream}.
+ */
+@Internal
+@Immutable
+final class BufferingInputStreamExtension implements InputStreamExtension {
+
+    private final InputStreamOpener opener;
+    private final int readBufferSize;
+
+    BufferingInputStreamExtension(final InputStreamOpener opener, final int readBufferSize) {
+        this.opener = Preconditions.checkNotNull(opener, "opener");
+        Preconditions.checkArgument(readBufferSize > 0, "readBufferSize must be positive");
+        this.readBufferSize = readBufferSize;
+    }
+
+    @Override
+    public RawAndWrappedInputStreams openStream(final InputStreamExtension.StreamContext ctx)
+            throws IOException {
+        final InputStream raw = opener.open(ReadContext.of(ctx.getPos()));
+        return new RawAndWrappedInputStreams(raw, new BufferedInputStream(raw, readBufferSize));
+    }
+}

flink-core/src/main/java/org/apache/flink/core/fs/InputStreamExtension.java

@@ -0,0 +1,74 @@
+/*
+ * 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.core.fs;
+
+import org.apache.flink.annotation.Internal;
+
+import java.io.IOException;
+
+/**
+ * Extension point for object-storage input stream implementations.
+ *
+ * <p>All methods are called with the stream's internal lock held. Implementations do not need
+ * their own synchronisation.
+ */
+@Internal
+public interface InputStreamExtension {
+
+    /** Read-only view of stream state passed to extension callbacks. */
+    @Internal
+    interface StreamContext {
+
+        /** Returns the current byte position. */
+        long getPos();
+
+        /** Returns the total content length in bytes. */
+        long getContentLength();
+    }
+
+    /**
+     * Opens streams at the current {@linkplain StreamContext#getPos() read position}.
+     *
+     * <p>Called during lazy initialization (first read) and stream recovery (seek beyond
+     * threshold). The previous streams are closed by the base class before this call. The base
+     * class manages the lifecycle of the returned streams (reads, closes, reopens on seek).
+     *
+     * <p>If the underlying source cannot do range reads (e.g., encrypted streams opened at offset
+     * 0), open at offset 0 and skip forward to {@link StreamContext#getPos()} after wrapping.
+     *
+     * <p>On failure, close any streams opened during this call before rethrowing.
+     *
+     * @param ctx read-only view of the stream state at the time of opening
+     * @return the raw and wrapped stream pair
+     * @throws IOException if opening fails
+     */
+    RawAndWrappedInputStreams openStream(StreamContext ctx) throws IOException;
+
+    /**
+     * Returns the default extension that opens the raw stream via {@code opener} and wraps it with
+     * a {@link java.io.BufferedInputStream}.
+     *
+     * @param opener opens a raw stream at a given byte position
+     * @param readBufferSize the buffer size for the {@link java.io.BufferedInputStream}
+     */
+    static InputStreamExtension buffering(
+            final InputStreamOpener opener, final int readBufferSize) {
+        return new BufferingInputStreamExtension(opener, readBufferSize);
+    }
+}

flink-core/src/main/java/org/apache/flink/core/fs/InputStreamOpener.java

@@ -0,0 +1,52 @@
+/*
+ * 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.core.fs;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.annotation.Internal;
+
+import javax.annotation.concurrent.NotThreadSafe;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * Cloud-agnostic opener that returns an input stream starting at the position described by a {@link
+ * ReadContext}.
+ *
+ * <p>The implementation captures all cloud-specific state (client, path, bucket) in its closure,
+ * exposing only the read context to the caller.
+ *
+ * @see ReadContext
+ */
+@Internal
+@Experimental
+@NotThreadSafe
+@FunctionalInterface
+public interface InputStreamOpener {
+
+    /**
+     * Opens an input stream starting at the position indicated by {@code ctx}.
+     *
+     * @param ctx context for this read, including the byte offset to start from
+     * @return an input stream positioned at {@code ctx.getPos()}
+     * @throws IOException if the stream cannot be opened
+     */
+    InputStream open(ReadContext ctx) throws IOException;
+}

flink-core/src/main/java/org/apache/flink/core/fs/ObjectStorageFileSystem.java

@@ -0,0 +1,77 @@
+/*
+ * 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.core.fs;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.util.CloseableIterator;
+
+import java.io.IOException;
+
+/**
+ * Optional capability for object storage filesystems that support rich metadata access and
+ * ETag-guarded atomic renames.
+ *
+ * <p>Implementations expose per-object metadata (e.g., encryption headers) and a conditional
+ * rename primitive used by re-encryption workflows. Discovered via {@code instanceof}.
+ *
+ * @see RichFileStatus
+ */
+@Internal
+@Experimental
+public interface ObjectStorageFileSystem {
+
+    /**
+     * Lazily enumerates all file paths under {@code prefix}, recursively.
+     *
+     * <p>The returned iterator must be closed to release any paging cursors or SDK connections.
+     *
+     * @param prefix the path prefix to enumerate; must not be {@code null}
+     * @return a closeable, lazy iterator over all file paths under {@code prefix}
+     * @throws IOException if enumeration cannot be started
+     */
+    CloseableIterator<Path> pathsList(Path prefix) throws IOException;
+
+    /**
+     * Fetches file status with user metadata and ETag populated.
+     *
+     * <p>Performs one HEAD-equivalent request per call.
+     *
+     * @param path the path to query; must not be {@code null}
+     * @return file status enriched with user metadata and ETag
+     * @throws IOException if the status cannot be fetched
+     */
+    RichFileStatus getRichFileStatus(Path path) throws IOException;
+
+    /**
+     * ETag-guarded atomic rename: moves {@code src} to {@code dst} only if the ETags match.
+     *
+     * <p>The move succeeds only when the source ETag matches {@code srcETag} and, if {@code
+     * dstETag} is non-empty, the destination ETag matches {@code dstETag}. An empty {@code dstETag}
+     * asserts that the destination does not exist.
+     *
+     * @param src source path; must not be {@code null}
+     * @param dst destination path; must not be {@code null}
+     * @param srcETag expected ETag of the source object; must not be {@code null}
+     * @param dstETag expected ETag of the destination object, or {@code ""} to assert non-existence
+     * @return {@code true} if the move succeeded; {@code false} if any ETag guard fired
+     * @throws IOException if the storage operation itself failed and the source state is unknown
+     */
+    boolean moveVerified(Path src, Path dst, String srcETag, String dstETag) throws IOException;
+}

flink-core/src/main/java/org/apache/flink/core/fs/OutputStreamOpener.java

@@ -0,0 +1,50 @@
+/*
+ * 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.core.fs;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.annotation.Internal;
+
+import javax.annotation.concurrent.NotThreadSafe;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * Cloud-agnostic opener that creates an output stream for a write operation.
+ *
+ * <p>The implementation captures all cloud-specific state (client, path, bucket) in its closure.
+ * The {@link WriteContext} provides encryption metadata to attach to the cloud object before the
+ * stream is opened.
+ */
+@Internal
+@Experimental
+@NotThreadSafe
+@FunctionalInterface
+public interface OutputStreamOpener {
+
+    /**
+     * Opens an output stream for the write operation described by {@code ctx}.
+     *
+     * @param ctx context for this write, including metadata to attach to the cloud object
+     * @return an output stream for writing the file content
+     * @throws IOException if the stream cannot be opened
+     */
+    OutputStream open(WriteContext ctx) throws IOException;
+}