Hello, I'm setting up a flink application to incrementally process files saved in s3 using FileSource API;

FileSource.forRecordStreamFormat(new Format(), new Path(path))
                .monitorContinuously(Duration.ofMinutes(15))
                .build()

In production, my job fails to start, despite increasing timeouts to 10 minutes, with one of the last lines in the logs being

org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - Starting split enumerator for source Source: S3.

Having debugged a bit locally, with a much smaller dataset, I believe the problem to be that for SplitEnumerator to start, it wants to first list all existing paths in S3 subdirectory. (of which there are many, 100k+ subdirectories) Is it possible to configure the enumerator to incrementally discover new s3 files, aka, allowing it to start without fully discovering entire s3 structure? Alternatively, can anyone share their experience with going around this problem?

I don’t know of a out of box config option to make the FileSource enumerate splits incrementally, but there are some ways to address this. You could implement a custom SplitEnumerator that is tailored to your use case. Extend SplitEnumerator class and implement logic that fetches and processes file paths incrementally instead of trying to fetch all at once. This way, you can control how the splits are discovered and loaded into Flink. This could be done by fetching a batch of paths, processing them, then fetching the next batch.