[FLINK-10518] Inefficient design in ContinuousFileMonitoringFunction

• Type: Improvement
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Affects Version/s: 1.5.2
• Component/s: Connectors / FileSystem
• Labels: Source:FileSystem auto-deprioritized-major auto-deprioritized-minor auto-unassigned
• Assignee: Unassigned
• Reporter: Huyen Levan
• Created: 09/Oct/18 11:39
• Updated: 26/Dec/21 22:39

Description

The ContinuousFileMonitoringFunction class keeps track of the latest file modification time to rule out all files it has processed in the previous cycles. For a long-running job, the list of eligible files will be much smaller than the list of all files in the folder being monitored.
In the current implementation of the getInputSplitsSortedByModTime method, a (big) list of all available splits are created first, and then every single split is checked with the list of eligible files.

for (FileInputSplit split: format.createInputSplits(readerParallelism)) {
FileStatus fileStatus = eligibleFiles.get(split.getPath());
if (fileStatus != null) {

The improvement can be done as:

• Listing of all files should be done once in ContinuousFileMonitoringFunction.listEligibleFiles() (as of now it is done the 2nd time in FileInputFormat.createInputSplits() )
• The list of file-splits should then be created from the list of paths in eligibleFiles.