zhangyue19921010 commented on code in PR #6133: URL: https://github.com/apache/hudi/pull/6133#discussion_r1028264254
########## hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerCheckerRunnable.java: ########## @@ -0,0 +1,193 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hudi.timeline.service.handlers.marker; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hudi.common.model.HoodieCommitMetadata; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.util.HoodieTimer; +import org.apache.hudi.common.util.MarkerUtils; +import org.apache.hudi.exception.HoodieIOException; +import org.apache.hudi.timeline.service.handlers.MarkerHandler; +import org.apache.log4j.LogManager; +import org.apache.log4j.Logger; + +import java.io.IOException; +import java.util.List; +import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +public class MarkerCheckerRunnable implements Runnable { + private static final Logger LOG = LogManager.getLogger(MarkerCheckerRunnable.class); + + private MarkerHandler markerHandler; + private String markerDir; + private String basePath; + private FileSystem fs; + private AtomicBoolean hasConflict; + private long maxAllowableHeartbeatIntervalInMs; + private Set<HoodieInstant> oldInstants; + + public MarkerCheckerRunnable(AtomicBoolean hasConflict, MarkerHandler markerHandler, String markerDir, + String basePath, FileSystem fileSystem, long maxAllowableHeartbeatIntervalInMs, + Set<HoodieInstant> oldInstants) { + this.markerHandler = markerHandler; + this.markerDir = markerDir; + this.basePath = basePath; + this.fs = fileSystem; + this.hasConflict = hasConflict; + this.maxAllowableHeartbeatIntervalInMs = maxAllowableHeartbeatIntervalInMs; + this.oldInstants = oldInstants; + } + + @Override + public void run() { + try { + if (!fs.exists(new Path(markerDir))) { + return; + } + + HoodieTimer timer = new HoodieTimer().startTimer(); + Set<String> currentInstantAllMarkers = markerHandler.getAllMarkers(markerDir); + Path tempPath = new Path(basePath + Path.SEPARATOR + HoodieTableMetaClient.TEMPFOLDER_NAME); + + List<Path> instants = MarkerUtils.getAllMarkerDir(tempPath, fs); + List<String> candidate = getCandidateInstants(instants, markerDirToInstantTime(markerDir)); + Set<String> tableMarkers = candidate.stream().flatMap(instant -> { + return MarkerUtils.readTimelineServerBasedMarkersFromFileSystemLocally(instant, fs).stream(); + }).collect(Collectors.toSet()); + + Set<String> currentFileIDs = currentInstantAllMarkers.stream().map(this::makerToPartitionAndFileID).collect(Collectors.toSet()); + Set<String> tableFilesIDs = tableMarkers.stream().map(this::makerToPartitionAndFileID).collect(Collectors.toSet()); + + currentFileIDs.retainAll(tableFilesIDs); + + if (!currentFileIDs.isEmpty() || hasCommitConflict(currentInstantAllMarkers, basePath)) { + LOG.warn("Conflict writing detected based on markers!\n" + + "Conflict markers: " + currentInstantAllMarkers + "\n" + + "Table markers: " + tableMarkers); + hasConflict.compareAndSet(false, true); + } + LOG.info("Finish batch marker checker in " + timer.endTimer() + " ms"); + + } catch (IOException e) { + throw new HoodieIOException("IOException occurs during checking marker conflict"); + } + } + + /** + * Get Candidate Instant to do conflict checking: + * 1. Skip current writer related instant(currentInstantTime) + * 2. Skip all instants after currentInstantTime + * 3. Skip dead writers related instants based on heart-beat + * @param instants + * @return + */ + private List<String> getCandidateInstants(List<Path> instants, String currentInstantTime) { Review Comment: Yeap, actually there are some diff here: for occ getCandidateInstants which depends on a state: ``` // To find which instants are conflicting, we apply the following logic // 1. Get completed instants timeline only for commits that have happened since the last successful write. // 2. Get any scheduled or completed compaction or clustering operations that have started and/or finished // after the current instant. We need to check for write conflicts since they may have mutated the same files // that are being newly created by the current write. ``` For current early conflict detection getCandidateInstants: ``` /** * Get Candidate Instant to do conflict checking: * 1. Skip current writer related instant(currentInstantTime) * 2. Skip all instants after currentInstantTime * 3. Skip dead writers related instants based on heart-beat * @param instants * @return */ ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
