Hi Samrat, Thank you for looking into FLIP-597[1] in great detail. And especially for trying it out for GCS/AWS implementations as this is the best way to find early issues with the interface design.
Please find my thoughts below: 1. Scope and interface layering *re: layers*: I can probably add a diagram to explain the layers and the flow. Explicitly calling out 3 layers makes sense to me. *re: handle*: Speaking of the interfaces, could you please elaborate on your proposal? As per my current understanding we can discuss the following tradeoffs: a. ReadContext, StreamContext, WriteContext Currently, they wrap only small subset of arguments to pass: ReadContext -> long (position to re-open) StreamContext -> long, long (position, contentLength) WriteContext -> map<string, string> (file metadata) Yes, these wraps could be inlined but the values I see in them are: more explicit contract; better symmetry in the interfaces. From a maintainability point of view, it seems to me that making changes to these wrappers is easier than tracing free passed arguments. Do you have performance concerns around them? As for InputStreamOpener, InputStreamExtension, and RawAndWrappedInputStreams, they all have distinct roles: the opener is the provider-specific SDK call, the extension for cross-cutting concerns like CSE[2], and the stream pair to separate the readable stream from SDK connection. We could in theory collapse some of them, but imho it would hurt testability and contract separation. Could you please provide more details on how to better outline contracts? I would also like to understand the reason for placing the implementations > in > flink-core Main motivation as mentioned in rejected alternatives section [3] is to avoid introduction of yet another dependency that every plugin would have to declare alongside core. It made more sense to me to put new interfaces alongside `FileSystem / FSDataInputStream`. 2. Thread safety and cancellation. > locking every read affected the throughput That is a slightly surprising finding. Given how we use streams in flink, access to the reentrant lock would be 99% uncontended (except the cancellation case). And most reads would be 8KB reads for the download path. I would imagine that unless you issue single byte reads, latency for acquiring uncontended locks should be negligible relative to the HTTP latency on the SDK side (multiple orders of magnitude smaller). Do you have more details on a benchmark scenario in which it mattered? during a blocking SDK read IIUC, SDK reads internally are non-interruptible. If you are concerned about locking in `close` non-interruptibly vs interruptibly elsewhere, I don't think It makes a big difference in terms of a hang risk. When stream is closed gracefully, it is closed from the same thread, hence it can't be stuck on the SDK read. So this hang could only ever have an impact during task cancellation. Which is asynchronous and doesn't prevent job re-submission. It would also be limited to SDK timeout, which is typically well below default timeout for task cancellation before the forceful slot release. If full thread safety is required Thread safety is required for CSE. The issue stems from the fact that CSE requires verification of read stream, which requires reading until the GCM verification tag. Unless/until we implement chunked encryption, we would have to "read input stream fully" to ensure that we can trust the data. Given that we need to read the stream on close, it requires full thread safety on reads between main reader and closer. There are only 2 ways I see that could avoid locking on reads: a. We implement chunked encryption CSE (worse performance on sequential reads, but much better random access, which is normally not used in flink). With chunked encryption, we could "read ahead" and only serve already verified data and eliminate the need to "read on close". b. We clearly separate graceful close (read data is used) from abortion (read data is never used). Abortion case would not require thread safety. And graceful close has to be called from the same thread. 3. Stream ownership, CRT, and encryption > S3 may require > aborting a response rather than closing and draining it Draining is only implemented in the CSE extension itself, it doesn't happen for standard buffering implementation. The only difference is the potential cancellation delay mentioned above if we had a hanging/stuck read. RawAndWrappedInputStreams may therefore need more > explicit rules for ownership, close ordering, and abort behaviour. Could you please provide additional requirements regarding the abort behaviour you have mentioned? This might let each provider adapt its native stream > without introducing an intermediate buffer The default buffering was introduced to provide forward seek optimisation. If you do forward skips within the block that is already pulled by SDK, it might entirely ignore the fact that you have data locally and instead open a new range read with the cloud provider. Having a buffering extension allows faster in-memory buffer drain instead of extra HTTP requests. If your concern is that you would like to skip buffering, we could add another "no-op" implementation, that essentially gives you a RawAndRaw input stream. In combination with idempotent `close`, I don't see it causing any issues. Random access will require more than metadata > injection For the client side encryption, the largest question is whether we have to go for chunked implementation already or not. Flink FS hot paths don't involve "random access". Files are read from beginning to the end, hence the benefits of encrypting small "chunks" vs the whole file is marginal. Standard format for algorithms like AES-GCM would be [Initialization Vector(IV)][Ciphertext][Verification Tag(GCM)]. In case of chunked encryption, it would be [IV1][Chunk1][GCM1][IV2][Chunk2][GCM2] etc. Initial implementation I have suggested in FLIP-601 [2] goes for the whole file encryption approach as a simpler and slightly faster first implementation. However, if you want to go for chunked encryption, we don't need much extra. Metadata injection gives you "chunk size/IV length/GCM tag length", then remaining work is to implement a couple of position mappers that would track reopen stream not from 0 position but "beginning of the closest chunk". In addition, for chunked encryption, we would swap "drain" for "read ahead". I will share the header-injection > and range-read requirements once that discussion concludes. Sounds great, I suggest we move this requirements discussion in FLIP-601 thread. 4. Existing open questions only shared utilities and contract tests? This part, but as we have discussed, cloud specific parts might be too diverse around recoverable writers, so juice probably not worth the squeeze and I would be happy to keep this one out of the scope. For PathsCopyingFileSystem, I believe the existing interface already > provides the common capability In general, I agree. The reason I mention it is that we have provided this optimisation for AWS in many different ways, but we don't really have it supported for other clouds, and I was wondering about where the value of the implementation sits. If it is heavily re-using SDK capabilities with thin logic, then yes, doing it per-cloud makes sense. However, if the added value is in handling concurrency over simple SDK calls, that could be largely cloud agnostic and be easily re-used for other clouds too. Having said that, I would suggest considering this as a follow up improvement rather than the main FLIP scope. demonstrate a stable intersection I am heavy +1 to that, let's see how it looks for all 3 and if there is a lot of code duplication, let's extract it. If the intended interface is narrower Intended interface is certainly narrower than all ObjectStorage operations and is naturally limited to operations that FS implementation demands. I have example of such implementation for Azure, and we use S3AccessHelper [4] For Native S3, most likely, these 2 will already be very representative, but if we can analyse what would be needed for GCS too, it would be helpful. 5. Exception classification Is your proposal to add a special exception wrapper interface that would intercept all FS exceptions and re-wrap them with pre-classification? It makes sense to me to have per-cloud classifier, but I am not sure where would this interface live and when would we re-wrap the exceptions. Would it be some FS wrapper like ExceptionClassifingFileSystem? If so, I would suggest having it as a followup/child FLIP. Could you please suggest how this classification would integrate with FS? 6. Migration I assume deprecation will be > evaluated separately for each cloud plugin It makes sense to consider deprecation timelines "per plugin". This FLIP focuses on hadoop-less flink filesystems, and we are on track to provide native versions for all main clouds. Hence I called it out as our general intention. There are more Hadoop uses in the Flink codebase but we are not aiming to address all of them in one go. We should also document that in-progress RecoverableWriter > State may not be portable between Hadoop and native implementations This is an important callout, it might block migration of jobs with object storage sinks. Do you have a detailed writeup on the challenges there that I could use as a reference? Thank you again for taking time to review and analyse the proposal. I look forward to converging this discussion to unblock further work for GCS and Azure support. Kind regards, Aleksandr Iushmanov References: [1] FLIP-597: Hadoop-less Flink Filesystems — https://cwiki.apache.org/confluence/x/9gDuGQ [2] FLIP-601: Client-Side Encryption Extension — https://cwiki.apache.org/confluence/x/P4E_Gg [3] https://cwiki.apache.org/confluence/spaces/FLINK/pages/435028214/FLIP-597+Filesystems+Hadoop-less+Flink+filesystems#FLIP597%3A%5BFilesystems%5DHadooplessFlinkfilesystems-Separateflink-fs-commonmodule [4] S3AccessHelper: https://github.com/apache/flink/blob/master/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3AccessHelper.java On Sun, 12 Jul 2026 at 06:44, Samrat Deb <[email protected]> wrote: > Hi Aleksandr, > > I have been applying the proposed interfaces to the native GCS prototype. > I > encountered a few questions that I would like to understand better. Some > may > come from gaps in my prototype or my understanding of the intended > layering, > so I would appreciate corrections and guidance. > > 1. Scope and interface layering > > Would it be helpful for the FLIP to show three explicit layers? > > - Existing Flink contracts, such as FileSystem and RecoverableWriter > - Shared object-storage mechanics such as position tracking and ranged > reopening > - Provider-specific SDK adapters and capabilities > > My current impression is that ReadContext, StreamContext, > InputStreamOpener, > InputStreamExtension, and RawAndWrappedInputStreams overlap somewhat. > Could we > initially reduce this to a position-aware stream factory or handle, and > add > further context only when a second use case requires it? > > I would also like to understand the reason for placing the > implementations in > flink-core rather than a small flink-object-storage-fs-base module. There > may > be plugin classloading constraints that I have missed. > > 2. Thread safety and cancellation > > In the GCS prototype, locking every read affected the throughput. There > is also a > Cancellation concern: when a lock is held during a blocking SDK read, a > Concurrent close cannot acquire the lock to abort that operation. > > Would a narrower contract be sufficient: reads and writes remain thread- > Is confined, while close or abort is safe to invoke concurrently during > task > cancellation? If full thread safety is required, it would help to > document the > Flink caller that requires it. > > I will share the GCS benchmark results and a blocked-read cancellation > test so > that we can evaluate this with data rather than assumptions. > > 3. Stream ownership, CRT, and encryption > > I understand the need to support normal SDK streams, AWS CRT, buffering, > and > future client-side encryption through one model. > > The remaining question for me is lifecycle ownership. S3 may require > aborting > a response rather than closing and draining it, while GCS and Azure > expose > different channel semantics. RawAndWrappedInputStreams may therefore need > more > explicit rules for ownership, close ordering, and abort behaviour. > > Could the opener return an owned handle containing the readable stream > and its > abort/close operation? This might let each provider adapt its native > stream > without introducing an intermediate buffer. I will test this approach > with > both the GCS prototype and the native S3 implementation. > > For client-side encryption, I am still gathering requirements from the > security discussion. Random access will require more than metadata > injection > because logical offsets, encrypted chunk boundaries, integrity tags, and > physical object lengths must be considered. I will share the > header-injection > and range-read requirements once that discussion concludes. > > 4. Existing open questions > > For RecoverableWriter, my current understanding is that RecoverableWriter > itself is already the cloud-neutral abstraction, while the recovery > protocol > should remain provider-specific. S3 uses multipart state, whereas the > existing > GCS writer uses immutable component objects and composes so that older > checkpoint handles remain valid after writing continues. Did you envisage > a > new common implementation here, or only shared utilities and contract > tests? > > For PathsCopyingFileSystem, I believe the existing interface already > provides > the common capability. Each provider could implement it independently > when an > optimised transfer path is available. > > For ObjectStorageOperations, I understand the motivation, especially > given the > declining availability of reliable emulators. One possible incremental > approach would be to use provider-local SDK adapters initially, and share > FileSystem behaviour tests, and extract a common operations interface > once S3, > GCS and Azure demonstrate a stable intersection. If the intended > interface is > narrower than a complete object-store abstraction, documenting the exact > operations and semantics would help me evaluate it against GCS. > > 5. Exception classification > > I am interested in connecting this work with the exception-classification > mechanism. It would be useful to align common categories such as > authentication, not-found, precondition failure, throttling, timeout, and > cancellation while retaining the original SDK exception as the cause. > > Should this classification be part of FLIP-597, or should the filesystem > preserve provider exceptions and let the classification plugin handle > them in > a follow-up? I would be happy to prepare a comparison of the AWS, GCS, > and > Azure exception models. > > One final clarification concerns migration. I assume deprecation will be > evaluated separately for each cloud plugin rather than applying to > flink-fs- > hadoop-shaded generally, because HDFS and the remaining Hadoop-based > integrations > still require it. We should also document that in-progress > RecoverableWriter > State may not be portable between Hadoop and native implementations. > > I will follow up with the GCS prototype, performance notes, the S3/GCS > behaviour comparison, and the encryption requirements. I am also happy to > help > refine the interface documentation or add shared contract tests. > > These questions are intended to help us arrive at the smallest > abstraction > that preserves correctness and performance across providers. I support > the > Hadoop-less direction, and I am keen to learn from the Azure and S3 > perspectives > so that we can help finalise FLIP-597. > > Bests, > Samrat > > On Thu, Jun 25, 2026 at 4:39 PM Aleksandr Iushmanov <[email protected]> > wrote: > > > Hi everyone, > > > > I'd like to start a discussion on FLIP-597: Hadoop-less Flink Filesystems > > [1]. > > > > As FLIP-555 [2] established, Flink's Hadoop-based filesystem plugins > > carry significant maintenance burdens: transitive dependency conflicts, > > classpath issues, and CVE exposure. FLIP-555 addressed this for S3 by > > introducing a native SDK-based filesystem. This FLIP extends that effort > > as an umbrella for all cloud providers (Azure, GCS, and potentially > > others) with the end goal to deprecate Hadoop-based variants. > > > > The FLIP proposes common, cloud-agnostic I/O abstractions placed in > > flink-core that all SDK-native filesystem implementations share as > > a step towards Hadoop-less Flink deployments: > > > > - ObjectStorageInputStream / ObjectStorageOutputStream: thread-safe > > stream implementations. > > - InputStreamExtension / InputStreamOpener: composable read pipeline > > supporting buffering, decryption via decoration > > - WriteContext / ReadContext: metadata descriptors enabling features > > like client-side encryption without changing the FileSystem API > > > > There are a number of open questions for discussion: > > 1. Should a cloud-agnostic RecoverableWriter abstraction be part of > > this FLIP given that implementation across clouds may be too specific > > for common abstractions to be useful. > > 2. Should PathsCopyingFileSystem (bulk copy) be in scope? Transfer > managers > > for clouds vary. And currently we only support this implementation in > > AWS. > > 3. Should this FLIP include an ObjectStorageOperations interface > > (analogous to FLIP-555's S3AccessHelper) for a generalized > > ObjectStorageFileSystem base, or should we defer premature > > generalisation > > of testing setup? Main motivation for this would be the ability to > > verify > > common FS parts with testing doubles with reduced reliance on other > > libraries > > like MinIO [3] (archived), LocalStack [4] (archived), Azurite [5] (no > > SDK V2 support) > > that demonstrates a decline in support. > > > > Looking forward to your feedback. > > > > Best regards, > > Aleksandr Iushmanov > > > > [1] https://cwiki.apache.org/confluence/x/9gDuGQ > > [2] https://cwiki.apache.org/confluence/x/uYqmFw > > [3] https://github.com/minio/minio > > [4] https://github.com/localstack/localstack > > [5] https://github.com/Azure/Azurite > > >
