Parallelize the process of constructing `logFilesMarkerPath` in CommitMetadatautils#reconcileMetadataForMissingFiles

This is related to HUDI-1517.
Current logic is:
{code:java}
Set logFilesMarkerPath = new HashSet<>();
allLogFilesMarkerPath.stream().filter(logFilePath -> !logFilePath.endsWith("cdc")).forEach(logFilesMarkerPath::add);

// remove valid log files
// TODO: refactor based on HoodieData
for (Map.Entry> partitionAndWriteStats : commitMetadata.getPartitionToWriteStats().entrySet()) {
for (HoodieWriteStat hoodieWriteStat : partitionAndWriteStats.getValue()) {
logFilesMarkerPath.remove(hoodieWriteStat.getPath());
}
} {code}
The for loop can be achieved via context.parallelize as below, but need to check for thread-safety.
{code:java}
Set logFilesMarkerPath = new HashSet<>();
allLogFilesMarkerPath.stream().filter(logFilePath -> !logFilePath.endsWith("cdc")).forEach(logFilesMarkerPath::add);

// Convert the partition and write stats to a list of log file paths to be removed
List validLogFilePaths = context.parallelize(new ArrayList<>(commitMetadata.getPartitionToWriteStats().entrySet()))
.flatMapToPair((SerializablePairFunction>, String, Void>) entry -> {
List> pathsToRemove = new ArrayList<>();
entry.getValue().forEach(hoodieWriteStat -> pathsToRemove.add(Pair.of(hoodieWriteStat.getPath(), null)));
return pathsToRemove.iterator();
})
.map(t -> t.getLeft())
.collect();

// Remove the valid log file paths from logFilesMarkerPath in a parallel manner
// Depending on the specifics of your environment and HoodieEngineContext, this might need to be adapted.
// For a straightforward approach without parallelization of the remove operation:
validLogFilePaths.forEach(logFilesMarkerPath::remove); {code}
 

## JIRA info

- Link: https://issues.apache.org/jira/browse/HUDI-7420
- Type: Task
- Fix version(s):
- 1.1.0