Post 2: Decide on a Solution Path

May 4, 2026 · Part of the Contributing to Apache Flink series

To reiterate the problem: for unbounded file sources, the FileEnumerator currently remembers the paths of every file it has processed, and that state can grow rather large. I'm looking into possibilities to reduce this state. Stated more technically, in the ContinuousFileSplitEnumerator the pathsAlreadyProcessed HashSet stored in JM state grows unboundedly and causes OOM issues.

From the comments on the Jira ticket (FLINK-25672), it's also stated that the problem is exacerbated by the fact that the state isn't distributed.

The two methods I'm weighing:

Does this need a FLIP?

I believe this MIGHT require a FLIP because some of the affected classes such as PendingSplitsCheckpoint have a @PublicEvolving annotation on them. PendingSplitsCheckpointSerializer might need to get bumped from V1 to V2, and that changes what gets written to checkpoints. The Flink contributor guide says: “If a change is identified as a large or controversial change in the discussion on Jira, it might require a Flink Improvement Proposal (FLIP) or a discussion on the Dev mailing list to reach agreement and consensus.” I don't believe this change meets that bar, but I'll be consulting with David Radley and Martijn Visser.

FLIP is for major changes or enhancements, which I don't believe this to be.

What's actually changing for FLINK-25672

In FileSourceSplit, fileModificationTime is a per-split metadata field that's captured once at enumeration time from FileStatus#getModificationTime() and carried through the rest of the pipeline. On local disk it's java.io.File#lastModified() (which under the hood is a stat(2) syscall). On HDFS it's whatever the NameNode reports. On S3 it's the Last-Modified header of the object. Each FileSystem implementation supplies its own FileStatus subclass. This means, for our purposes, the modification timestamp already exists.

The current Java logic looks like this:

// state held in memory
private final HashSet<Path> pathsAlreadyProcessed;


// Filtering, called every discoveryInterval
private void processDiscoveredSplits(Collection<FileSourceSplit> splits, Throwable error) {
    if (error != null) {
        LOG.error("Failed to enumerate files", error);
        return;
    }

    final Collection<FileSourceSplit> newSplits =
        splits.stream()
                .filter((split) -> pathsAlreadyProcessed.add(split.path())) // this is the main line that causes paths to keep getting added, making the state grow
                .collect(Collectors.toList());
    splitAssigner.addSplits(newSplits);

    assignSplits();
}


public PendingSplitsCheckpoint<FileSourceSplit> snapshotState(long checkpointId)
        throws Exception {
    final PendingSplitsCheckpoint<FileSourceSplit> checkpoint =
            PendingSplitsCheckpoint.fromCollectionSnapshot(
                    splitAssigner.remainingSplits(), pathsAlreadyProcessed); // change needs to be made here as well

    LOG.debug("Source Checkpoint is {}", checkpoint);
    return checkpoint;
}

When it comes to HashSet<Path>, this works because if a path is in the set we've seen it before, and if it's not then we haven't. The assumption being made here is that the set is bounded by the user's needs. For a job watching a small directory of config files, or a job that runs for a few hours, this is true. But as we saw in Post 1, for a job running for months over a directory that grows by 100k files per day, the set is not bounded, and that's where issues arise.

My mindset in approaching this: there has to be a way to give users the choice of opting into the current behavior when they need a bounded representation, but at the same time, Flink is built for streaming and unbounded sources, so we have to make allowances for that as well.

Method 1: Timestamp watermark tracking

We know that FileSourceSplit carries the fileModificationTime:

/** Returns the modification time of the file, from {@link FileStatus#getModificationTime()}. */
public long fileModificationTime() {
    return fileModificationTime;
}

If we assume that files appear in the directory in roughly modification-time order, then once we've processed every file with modTime <= currentTime we no longer need to remember any individual paths older than currentTime. We do need to remember currentTime itself, plus those files whose paths sit right at the boundary.

The HashSet<Path> should be replaced with two lines:

long highestModTime;
final HashSet<Path> recentPaths;

I'll add a dedupe decision and check it against a small set. After processing a discovery batch, advance the watermark.

Concerns: it's possible for a producer to write a file with a modTime older than the current watermark minus the margin, and if that's the case the file gets silently skipped. This can happen if:

In my mind, anything could happen to cause out-of-order modification times, and that's a concern of mine.

Method 2: TTL-based eviction

The core idea here is to keep the same HashSet<Path> semantics, but every entry also carries a timestamp (what time was this file processed), and entries older than a configured retention duration get evicted. This method relies on the user having some mechanism that ensures files older than the agreed-upon retention time won't reappear in the directory listing. Think something like an S3 lifecycle rule.

private final LinkedHashMap<Path, Long> processedPathsByTime;
private final Duration retentionDuration;

With this solution the state could still get huge. The retention time we're bound by could still result in a large state, but for a lot of users the bound matters more than the size. A bound of “7 days of files” is something you can plan capacity around, even if it's still x entries. Today's behavior gives them no bound at all.

Cons:

Where I'm at

I'm not sure which of these solutions to go for, and I've sent an email to the Flink dev mailing list. I'm hoping to get feedback from others in the community before crafting a path forward for either method.

← Post 1: Problem FamiliarityNext up: Post 3, Understand the Codebase (coming soon)