steveloughran commented on a change in pull request #2971: URL: https://github.com/apache/hadoop/pull/2971#discussion_r824992486
########## File path: hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/committer/manifest/stages/CleanupJobStage.java ########## @@ -0,0 +1,561 @@ +/* + * 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.hadoop.mapreduce.lib.output.committer.manifest.stages; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.util.concurrent.atomic.AtomicInteger; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.RemoteIterator; +import org.apache.hadoop.mapreduce.lib.output.committer.manifest.impl.ManifestStoreOperations; +import org.apache.hadoop.util.DurationInfo; +import org.apache.hadoop.util.functional.RemoteIterators; +import org.apache.hadoop.util.functional.TaskPool; + +import static java.util.Objects.requireNonNull; +import static org.apache.hadoop.fs.statistics.IOStatisticsSupport.retrieveIOStatistics; +import static org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.FILEOUTPUTCOMMITTER_CLEANUP_FAILURES_IGNORED; +import static org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.FILEOUTPUTCOMMITTER_CLEANUP_FAILURES_IGNORED_DEFAULT; +import static org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.FILEOUTPUTCOMMITTER_CLEANUP_SKIPPED; +import static org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.FILEOUTPUTCOMMITTER_CLEANUP_SKIPPED_DEFAULT; +import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterConstants.OPT_CLEANUP_MOVE_TO_TRASH; +import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterConstants.OPT_CLEANUP_MOVE_TO_TRASH_DEFAULT; +import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterConstants.OPT_CLEANUP_PARALLEL_ATTEMPT_DIRS; +import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterConstants.OPT_CLEANUP_PARALLEL_ATTEMPT_DIRS_DEFAULT; +import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_STAGE_JOB_CLEANUP; + +/** + * Clean up a job's temporary directory through parallel delete, + * base _temporary delete and as a fallback, rename to trash. + * Returns: the outcome of the overall operation and any move to trash. + * The result is detailed purely for the benefit of tests, which need + * to make assertions about error handling and fallbacks. + */ +public class CleanupJobStage extends + AbstractJobCommitStage< + CleanupJobStage.Arguments, + CleanupJobStage.Result> { + + private static final Logger LOG = LoggerFactory.getLogger( + CleanupJobStage.class); + + /** + * Count of deleted directories. + */ + private final AtomicInteger deleteDirCount = new AtomicInteger(); + + /** + * Count of delete failures. + */ + private final AtomicInteger deleteFailureCount = new AtomicInteger(); + + /** + * Last delete exception; non null if deleteFailureCount is not zero. + */ + private IOException lastDeleteException = null; + + /** + * Stage name as passed in from arguments. + */ + private String stageName = OP_STAGE_JOB_CLEANUP; + + public CleanupJobStage(final StageConfig stageConfig) { + super(false, stageConfig, OP_STAGE_JOB_CLEANUP, true); + } + + /** + * Statistic name is extracted from the arguments. + * @param arguments args to the invocation. + * @return stage name. + */ + @Override + protected String getStageStatisticName(Arguments arguments) { + return arguments.statisticName; + } + + /** + * Clean up the job attempt directory tree. + * @param args arguments built up. + * @return the result. + * @throws IOException failure was raised an exceptions weren't surpressed. + */ + @Override + protected Result executeStage( + final Arguments args) + throws IOException { + stageName = getStageName(args); + // this is $dest/_temporary + final Path baseDir = requireNonNull(getStageConfig().getOutputTempSubDir()); + LOG.debug("{}: Cleaup of directory {} with {}", getName(), baseDir, args); + if (!args.enabled) { + LOG.info("{}: Cleanup of {} disabled", getName(), baseDir); + return new Result(Outcome.DISABLED, baseDir, + 0, null, null); + } + // shortcut of a single existence check before anything else + if (getFileStatusOrNull(baseDir) == null) { + return new Result(Outcome.NOTHING_TO_CLEAN_UP, + baseDir, + 0, null, null); + } + + // move to trash? + // this will be set if delete fails. + boolean moveToTrash = args.moveToTrash; Review comment: i tried to make this resilient to failures, but yes, it has got over complex. i'm going to roll back and be minimal. > If the FileSystem does have trash enabled - the delete FS operation would automatically take care of moving the contents to trash? doesn't actually. the move to trash code i have is lifted from the FsShell code, which is primarily where it gets used. that is: except through the command line, deleting files in the hadoop FS APIs doesn't use trash. -- 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: common-issues-unsubscr...@hadoop.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: common-issues-unsubscr...@hadoop.apache.org For additional commands, e-mail: common-issues-h...@hadoop.apache.org
