class DirectoryMonitorDiscovery implements PartitionDiscovery {

	@VisibleForTesting
	static List<Tuple2<List<String>, Long>> suitablePartitions(
			Context context,
			long previousTimestamp,
			FileStatus[] statuses) {
		List<Tuple2<List<String>, Long>> partValueList = new ArrayList<>();
		for (FileStatus status : statuses) {
			List<String> partValues = extractPartitionValues(
					new org.apache.flink.core.fs.Path(status.getPath().toString()));
			long timestamp = context.extractTimestamp(
					context.partitionKeys(),
					partValues,
					// to UTC millisecond.
					() -> TimestampData.fromTimestamp(
							new Timestamp(status.getModificationTime())).getMillisecond());
			if (timestamp >= previousTimestamp) {
				partValueList.add(new Tuple2<>(partValues, timestamp));
			}
		}
		return partValueList;
	}

}