vinothchandar commented on a change in pull request #1756:
URL: https://github.com/apache/hudi/pull/1756#discussion_r456965828
##########
File path: hudi-client/src/main/java/org/apache/hudi/io/HoodieWriteHandle.java
##########
@@ -97,28 +98,9 @@ public Path makeNewPath(String partitionPath) {
*
* @param partitionPath Partition path
*/
- protected void createMarkerFile(String partitionPath) {
- Path markerPath = makeNewMarkerPath(partitionPath);
- try {
- LOG.info("Creating Marker Path=" + markerPath);
- fs.create(markerPath, false).close();
- } catch (IOException e) {
- throw new HoodieException("Failed to create marker file " + markerPath,
e);
- }
- }
-
- /**
- * THe marker path will be
<base-path>/.hoodie/.temp/<instant_ts>/2019/04/25/filename.
- */
- private Path makeNewMarkerPath(String partitionPath) {
Review comment:
all of this stuff is now encapsulatd into a `MarkerFiles` class
##########
File path:
hudi-client/src/main/java/org/apache/hudi/table/HoodieTimelineArchiveLog.java
##########
@@ -264,6 +275,7 @@ public void archive(List<HoodieInstant> instants) throws
HoodieCommitException {
List<IndexedRecord> records = new ArrayList<>();
for (HoodieInstant hoodieInstant : instants) {
try {
+ deleteAnyLeftOverMarkerFiles(hoodieInstant);
Review comment:
during archival. either the commit instant or the corresponding
rollback.. any left over marker dir will be deleted. or the archival will fail.
there is a test added for this.
##########
File path: hudi-client/src/main/java/org/apache/hudi/table/HoodieTable.java
##########
@@ -410,72 +412,54 @@ public void deleteMarkerDir(String instantTs) {
* @param consistencyCheckEnabled Consistency Check Enabled
* @throws HoodieIOException
*/
- protected void cleanFailedWrites(JavaSparkContext jsc, String instantTs,
List<HoodieWriteStat> stats,
- boolean consistencyCheckEnabled) throws HoodieIOException {
+ protected void reconcileAgainstMarkers(JavaSparkContext jsc,
+ String instantTs,
+ List<HoodieWriteStat> stats,
+ boolean consistencyCheckEnabled)
throws HoodieIOException {
try {
// Reconcile marker and data files with WriteStats so that partially
written data-files due to failed
// (but succeeded on retry) tasks are removed.
String basePath = getMetaClient().getBasePath();
- FileSystem fs = getMetaClient().getFs();
- Path markerDir = new Path(metaClient.getMarkerFolderPath(instantTs));
+ MarkerFiles markers = new MarkerFiles(this, instantTs);
- if (!fs.exists(markerDir)) {
- // Happens when all writes are appends
+ if (!markers.doesMarkerDirExist()) {
+ // can happen if it was an empty write say.
return;
}
- final String baseFileExtension = getBaseFileFormat().getFileExtension();
- List<String> invalidDataPaths = FSUtils.getAllDataFilesForMarkers(fs,
basePath, instantTs, markerDir.toString(),
- baseFileExtension);
- List<String> validDataPaths = stats.stream().map(w ->
String.format("%s/%s", basePath, w.getPath()))
- .filter(p ->
p.endsWith(baseFileExtension)).collect(Collectors.toList());
+ // we are not including log appends here, since they are already
fail-safe.
Review comment:
quick skim of changes here would be nice.
##########
File path: hudi-client/src/main/java/org/apache/hudi/table/MarkerFiles.java
##########
@@ -0,0 +1,153 @@
+/*
+ * 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.table;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.io.IOType;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * Operates on marker files for a given write action (commit, delta commit,
compaction).
+ */
+public class MarkerFiles {
+
+ private static final Logger LOG = LogManager.getLogger(MarkerFiles.class);
+
+ public static String stripMarkerSuffix(String path) {
+ return path.substring(0, path.indexOf(HoodieTableMetaClient.MARKER_EXTN));
+ }
+
+ private final String instantTime;
+ private final FileSystem fs;
+ private final Path markerDirPath;
+ private final String basePath;
+
+ public MarkerFiles(FileSystem fs, String basePath, String markerFolderPath,
String instantTime) {
+ this.instantTime = instantTime;
+ this.fs = fs;
+ this.markerDirPath = new Path(markerFolderPath);
+ this.basePath = basePath;
+ }
+
+ public MarkerFiles(HoodieTable<?> table, String instantTime) {
+ this(table.getMetaClient().getFs(),
+ table.getMetaClient().getBasePath(),
+ table.getMetaClient().getMarkerFolderPath(instantTime),
+ instantTime);
+ }
+
+ public void quietDeleteMarkerDir() {
+ try {
+ deleteMarkerDir();
+ } catch (HoodieIOException ioe) {
+ LOG.warn("Error deleting marker directory for instant " + instantTime,
ioe);
+ }
+ }
+
+ /**
+ * Delete Marker directory corresponding to an instant.
+ */
+ public boolean deleteMarkerDir() {
Review comment:
@umehrot2 we can now add the parallelization changes here. for deletion
of marker files.
##########
File path:
hudi-client/src/main/java/org/apache/hudi/table/action/rollback/MarkerBasedRollbackStrategy.java
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.table.action.rollback;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.common.HoodieRollbackStat;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.table.log.HoodieLogFormat;
+import org.apache.hudi.common.table.log.block.HoodieCommandBlock;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.exception.HoodieRollbackException;
+import org.apache.hudi.io.IOType;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.MarkerFiles;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.apache.spark.api.java.JavaSparkContext;
+import scala.Tuple2;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Performs rollback using marker files generated during the write..
+ */
+public class MarkerBasedRollbackStrategy implements
BaseRollbackActionExecutor.RollbackStrategy {
+
+ private static final Logger LOG =
LogManager.getLogger(MarkerBasedRollbackStrategy.class);
+
+ private final HoodieTable<?> table;
+
+ private final transient JavaSparkContext jsc;
+
+ private final HoodieWriteConfig config;
+
+ private final String basePath;
+
+ private final String instantTime;
+
+ public MarkerBasedRollbackStrategy(HoodieTable<?> table, JavaSparkContext
jsc, HoodieWriteConfig config, String instantTime) {
+ this.table = table;
+ this.jsc = jsc;
+ this.basePath = table.getMetaClient().getBasePath();
+ this.config = config;
+ this.instantTime = instantTime;
+ }
+
+ private HoodieRollbackStat undoMerge(String mergedBaseFilePath) throws
IOException {
+ LOG.info("Rolling back by deleting the merged base file:" +
mergedBaseFilePath);
+ return deleteBaseFile(mergedBaseFilePath);
+ }
+
+ private HoodieRollbackStat undoCreate(String createdBaseFilePath) throws
IOException {
+ LOG.info("Rolling back by deleting the created base file:" +
createdBaseFilePath);
+ return deleteBaseFile(createdBaseFilePath);
+ }
+
+ private HoodieRollbackStat deleteBaseFile(String baseFilePath) throws
IOException {
Review comment:
this is resilient already to attempting to delete an non-existent file..
marker file may not imply the data file is thre.
##########
File path:
hudi-client/src/main/java/org/apache/hudi/client/HoodieWriteClient.java
##########
@@ -332,9 +333,12 @@ public static SparkConf registerClasses(SparkConf conf) {
}
@Override
- protected void postCommit(HoodieCommitMetadata metadata, String instantTime,
- Option<Map<String, String>> extraMetadata) {
+ protected void postCommit(HoodieTable<?> table, HoodieCommitMetadata
metadata, String instantTime, Option<Map<String, String>> extraMetadata) {
try {
+
+ // Delete the marker directory for the instant.
Review comment:
this PR will change behavior for marker dir deletion, with or without
marker based rollback turned on.
##########
File path:
hudi-client/src/main/java/org/apache/hudi/table/action/rollback/MarkerBasedRollbackStrategy.java
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.table.action.rollback;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.common.HoodieRollbackStat;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.table.log.HoodieLogFormat;
+import org.apache.hudi.common.table.log.block.HoodieCommandBlock;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.exception.HoodieRollbackException;
+import org.apache.hudi.io.IOType;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.MarkerFiles;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.apache.spark.api.java.JavaSparkContext;
+import scala.Tuple2;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Performs rollback using marker files generated during the write..
+ */
+public class MarkerBasedRollbackStrategy implements
BaseRollbackActionExecutor.RollbackStrategy {
+
+ private static final Logger LOG =
LogManager.getLogger(MarkerBasedRollbackStrategy.class);
+
+ private final HoodieTable<?> table;
+
+ private final transient JavaSparkContext jsc;
+
+ private final HoodieWriteConfig config;
+
+ private final String basePath;
+
+ private final String instantTime;
+
+ public MarkerBasedRollbackStrategy(HoodieTable<?> table, JavaSparkContext
jsc, HoodieWriteConfig config, String instantTime) {
+ this.table = table;
+ this.jsc = jsc;
+ this.basePath = table.getMetaClient().getBasePath();
+ this.config = config;
+ this.instantTime = instantTime;
+ }
+
+ private HoodieRollbackStat undoMerge(String mergedBaseFilePath) throws
IOException {
+ LOG.info("Rolling back by deleting the merged base file:" +
mergedBaseFilePath);
+ return deleteBaseFile(mergedBaseFilePath);
+ }
+
+ private HoodieRollbackStat undoCreate(String createdBaseFilePath) throws
IOException {
+ LOG.info("Rolling back by deleting the created base file:" +
createdBaseFilePath);
+ return deleteBaseFile(createdBaseFilePath);
+ }
+
+ private HoodieRollbackStat deleteBaseFile(String baseFilePath) throws
IOException {
+ Path fullDeletePath = new Path(basePath, baseFilePath);
+ String partitionPath = FSUtils.getRelativePartitionPath(new
Path(basePath), fullDeletePath.getParent());
+ boolean isDeleted = table.getMetaClient().getFs().delete(fullDeletePath);
+ return HoodieRollbackStat.newBuilder()
+ .withPartitionPath(partitionPath)
+ .withDeletedFileResult(baseFilePath, isDeleted)
+ .build();
+ }
+
+ private HoodieRollbackStat undoAppend(String appendBaseFilePath,
HoodieInstant instantToRollback) throws IOException, InterruptedException {
+ Path baseFilePathForAppend = new Path(basePath, appendBaseFilePath);
+ String fileId = FSUtils.getFileIdFromFilePath(baseFilePathForAppend);
+ String baseCommitTime =
FSUtils.getCommitTime(baseFilePathForAppend.getName());
+ String partitionPath = FSUtils.getRelativePartitionPath(new
Path(basePath), new Path(basePath, appendBaseFilePath).getParent());
+
+ HoodieLogFormat.Writer writer = null;
+ try {
+ Path partitionFullPath = FSUtils.getPartitionPath(basePath,
partitionPath);
+
+ if (!table.getMetaClient().getFs().exists(partitionFullPath)) {
+ return HoodieRollbackStat.newBuilder()
+ .withPartitionPath(partitionPath)
+ .build();
+ }
+ writer = HoodieLogFormat.newWriterBuilder()
Review comment:
checked that the log scanner can deal with spurious rollback blocks..
i.e rollbacks logged without any data blocks for that instant
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
