[
https://issues.apache.org/jira/browse/NUTCH-2518?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16433692#comment-16433692
]
ASF GitHub Bot commented on NUTCH-2518:
---------------------------------------
sebastian-nagel closed pull request #307: NUTCH-2518 Cleaning up the file
system after a job failure.
URL: https://github.com/apache/nutch/pull/307
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/src/java/org/apache/nutch/crawl/CrawlDb.java
b/src/java/org/apache/nutch/crawl/CrawlDb.java
index a5455099d..05fc3c67d 100644
--- a/src/java/org/apache/nutch/crawl/CrawlDb.java
+++ b/src/java/org/apache/nutch/crawl/CrawlDb.java
@@ -129,17 +129,23 @@ public void update(Path crawlDb, Path[] segments, boolean
normalize,
LOG.info("CrawlDb update: Merging segment data into db.");
}
+ FileSystem fs = crawlDb.getFileSystem(getConf());
+ Path outPath = FileOutputFormat.getOutputPath(job);
try {
- int complete = job.waitForCompletion(true)?0:1;
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "Crawl job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ NutchJob.cleanupAfterFailure(outPath, lock, fs);
+ throw new RuntimeException(message);
+ }
} catch (IOException | InterruptedException | ClassNotFoundException e) {
- FileSystem fs = crawlDb.getFileSystem(getConf());
- LockUtil.removeLockFile(fs, lock);
- Path outPath = FileOutputFormat.getOutputPath(job);
- if (fs.exists(outPath))
- fs.delete(outPath, true);
+ LOG.error("Crawl job failed {}", e);
+ NutchJob.cleanupAfterFailure(outPath, lock, fs);
throw e;
}
-
CrawlDb.install(job, crawlDb);
diff --git a/src/java/org/apache/nutch/crawl/CrawlDbMerger.java
b/src/java/org/apache/nutch/crawl/CrawlDbMerger.java
index 35eca6069..d8756fd5e 100644
--- a/src/java/org/apache/nutch/crawl/CrawlDbMerger.java
+++ b/src/java/org/apache/nutch/crawl/CrawlDbMerger.java
@@ -144,15 +144,23 @@ public void merge(Path output, Path[] dbs, boolean
normalize, boolean filter)
}
FileInputFormat.addInputPath(job, new Path(dbs[i],
CrawlDb.CURRENT_NAME));
}
+
+ Path outPath = FileOutputFormat.getOutputPath(job);
+ FileSystem fs = outPath.getFileSystem(getConf());
try {
- int complete = job.waitForCompletion(true)?0:1;
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "CrawlDbMerger job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ NutchJob.cleanupAfterFailure(outPath, lock, fs);
+ throw new RuntimeException(message);
+ }
CrawlDb.install(job, output);
- } catch (IOException e) {
- LockUtil.removeLockFile(getConf(), lock);
- Path outPath = FileOutputFormat.getOutputPath(job);
- FileSystem fs = outPath.getFileSystem(getConf());
- if (fs.exists(outPath))
- fs.delete(outPath, true);
+ } catch (IOException | InterruptedException | ClassNotFoundException e) {
+ LOG.error("CrawlDbMerge job failed {}", e);
+ NutchJob.cleanupAfterFailure(outPath, lock, fs);
throw e;
}
long end = System.currentTimeMillis();
diff --git a/src/java/org/apache/nutch/crawl/CrawlDbReader.java
b/src/java/org/apache/nutch/crawl/CrawlDbReader.java
index c1a79e991..dcf5acead 100644
--- a/src/java/org/apache/nutch/crawl/CrawlDbReader.java
+++ b/src/java/org/apache/nutch/crawl/CrawlDbReader.java
@@ -401,15 +401,23 @@ public void close() {
// https://issues.apache.org/jira/browse/NUTCH-1029
config.setBoolean("mapreduce.fileoutputcommitter.marksuccessfuljobs",
false);
-
+ FileSystem fileSystem = tmpFolder.getFileSystem(config);
try {
- int complete = job.waitForCompletion(true)?0:1;
- } catch (InterruptedException | ClassNotFoundException e) {
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "CrawlDbReader job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ fileSystem.delete(tmpFolder, true);
+ throw new RuntimeException(message);
+ }
+ } catch (IOException | InterruptedException | ClassNotFoundException
e) {
LOG.error(StringUtils.stringifyException(e));
+ fileSystem.delete(tmpFolder, true);
throw e;
}
// reading the result
- FileSystem fileSystem = tmpFolder.getFileSystem(config);
SequenceFile.Reader[] readers =
SegmentReaderUtil.getReaders(tmpFolder, config);
Text key = new Text();
@@ -684,8 +692,15 @@ public void processDumpJob(String crawlDb, String output,
job.setOutputValueClass(CrawlDatum.class);
try {
- int complete = job.waitForCompletion(true)?0:1;
- } catch (InterruptedException | ClassNotFoundException e) {
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "CrawlDbReader job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ throw new RuntimeException(message);
+ }
+ } catch (IOException | InterruptedException | ClassNotFoundException e) {
LOG.error(StringUtils.stringifyException(e));
throw e;
}
@@ -788,11 +803,21 @@ public void processTopNJob(String crawlDb, long topN,
float min,
job.setOutputValueClass(Text.class);
job.getConfiguration().setFloat("db.reader.topn.min", min);
-
+
+ FileSystem fs = tempDir.getFileSystem(config);
try{
- int complete = job.waitForCompletion(true)?0:1;
- } catch (InterruptedException | ClassNotFoundException e) {
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "CrawlDbReader job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ fs.delete(tempDir, true);
+ throw new RuntimeException(message);
+ }
+ } catch (IOException | InterruptedException | ClassNotFoundException e) {
LOG.error(StringUtils.stringifyException(e));
+ fs.delete(tempDir, true);
throw e;
}
@@ -816,13 +841,21 @@ public void processTopNJob(String crawlDb, long topN,
float min,
job.setNumReduceTasks(1); // create a single file.
try{
- int complete = job.waitForCompletion(true)?0:1;
- } catch (InterruptedException | ClassNotFoundException e) {
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "CrawlDbReader job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ fs.delete(tempDir, true);
+ throw new RuntimeException(message);
+ }
+ } catch (IOException | InterruptedException | ClassNotFoundException e) {
LOG.error(StringUtils.stringifyException(e));
+ fs.delete(tempDir, true);
throw e;
}
- FileSystem fs = tempDir.getFileSystem(config);
fs.delete(tempDir, true);
if (LOG.isInfoEnabled()) {
LOG.info("CrawlDb topN: done");
diff --git a/src/java/org/apache/nutch/crawl/DeduplicationJob.java
b/src/java/org/apache/nutch/crawl/DeduplicationJob.java
index f2283ee9c..555f9e2eb 100644
--- a/src/java/org/apache/nutch/crawl/DeduplicationJob.java
+++ b/src/java/org/apache/nutch/crawl/DeduplicationJob.java
@@ -312,8 +312,17 @@ public int run(String[] args) throws IOException {
job.setMapperClass(DBFilter.class);
job.setReducerClass(DedupReducer.class);
+ FileSystem fs = tempDir.getFileSystem(getConf());
try {
- int complete = job.waitForCompletion(true)?0:1;
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "Crawl job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ fs.delete(tempDir, true);
+ throw new RuntimeException(message);
+ }
CounterGroup g = job.getCounters().getGroup("DeduplicationJobStatus");
if (g != null) {
Counter counter = g.findCounter("Documents marked as duplicate");
@@ -321,8 +330,9 @@ public int run(String[] args) throws IOException {
LOG.info("Deduplication: " + (int) dups
+ " documents marked as duplicates");
}
- } catch (final Exception e) {
+ } catch (IOException | InterruptedException | ClassNotFoundException e) {
LOG.error("DeduplicationJob: " + StringUtils.stringifyException(e));
+ fs.delete(tempDir, true);
return -1;
}
@@ -337,16 +347,24 @@ public int run(String[] args) throws IOException {
mergeJob.setReducerClass(StatusUpdateReducer.class);
try {
- int complete = job.waitForCompletion(true)?0:1;
- } catch (final Exception e) {
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "Crawl job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ fs.delete(tempDir, true);
+ throw new RuntimeException(message);
+ }
+ } catch (IOException | InterruptedException | ClassNotFoundException e) {
LOG.error("DeduplicationMergeJob: " + StringUtils.stringifyException(e));
+ fs.delete(tempDir, true);
return -1;
}
CrawlDb.install(mergeJob, dbPath);
// clean up
- FileSystem fs = tempDir.getFileSystem(getConf());
fs.delete(tempDir, true);
long end = System.currentTimeMillis();
diff --git a/src/java/org/apache/nutch/crawl/Generator.java
b/src/java/org/apache/nutch/crawl/Generator.java
index c972a13c0..c9096d688 100644
--- a/src/java/org/apache/nutch/crawl/Generator.java
+++ b/src/java/org/apache/nutch/crawl/Generator.java
@@ -781,10 +781,18 @@ public Generator(Configuration conf) {
MultipleOutputs.addNamedOutput(job, "sequenceFiles",
SequenceFileOutputFormat.class, FloatWritable.class, SelectorEntry.class);
try {
- int complete = job.waitForCompletion(true)?0:1;
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "Generator job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ NutchJob.cleanupAfterFailure(tempDir, lock, fs);
+ throw new RuntimeException(message);
+ }
} catch (IOException | InterruptedException | ClassNotFoundException e) {
- LockUtil.removeLockFile(getConf(), lock);
- fs.delete(tempDir, true);
+ LOG.error("Generator job failed {}", e);
+ NutchJob.cleanupAfterFailure(tempDir, lock, fs);
throw e;
}
@@ -838,12 +846,21 @@ public Generator(Configuration conf) {
job.setOutputValueClass(CrawlDatum.class);
FileOutputFormat.setOutputPath(job, tempDir2);
try {
- int complete = job.waitForCompletion(true)?0:1;
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "Generator job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ NutchJob.cleanupAfterFailure(tempDir, lock, fs);
+ NutchJob.cleanupAfterFailure(tempDir2, lock, fs);
+ throw new RuntimeException(message);
+ }
CrawlDb.install(job, dbDir);
} catch (IOException | InterruptedException | ClassNotFoundException e) {
- LockUtil.removeLockFile(getConf(), lock);
- fs.delete(tempDir, true);
- fs.delete(tempDir2, true);
+ LOG.error("Generator job failed {}", e);
+ NutchJob.cleanupAfterFailure(tempDir, lock, fs);
+ NutchJob.cleanupAfterFailure(tempDir2, lock, fs);
throw e;
}
@@ -894,8 +911,15 @@ private Path partitionSegment(Path segmentsDir, Path
inputDir, int numLists)
job.setOutputValueClass(CrawlDatum.class);
job.setSortComparatorClass(HashComparator.class);
try {
- int complete = job.waitForCompletion(true)?0:1;
- } catch (InterruptedException | ClassNotFoundException e) {
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "Generator job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ throw new RuntimeException(message);
+ }
+ } catch (IOException | InterruptedException | ClassNotFoundException e) {
LOG.error(StringUtils.stringifyException(e));
throw e;
}
diff --git a/src/java/org/apache/nutch/crawl/Injector.java
b/src/java/org/apache/nutch/crawl/Injector.java
index d872dbff5..f73739ecf 100644
--- a/src/java/org/apache/nutch/crawl/Injector.java
+++ b/src/java/org/apache/nutch/crawl/Injector.java
@@ -44,6 +44,7 @@
import org.apache.nutch.service.NutchServer;
import org.apache.nutch.util.NutchConfiguration;
import org.apache.nutch.util.NutchTool;
+import org.apache.nutch.util.NutchJob;
import org.apache.nutch.util.TimingUtil;
import org.slf4j.Logger;
@@ -420,7 +421,7 @@ public void inject(Path crawlDb, Path urlDir, boolean
overwrite,
+ job.getStatus().getState() + ", reason: "
+ job.getStatus().getFailureInfo();
LOG.error(message);
- cleanupAfterFailure(tempCrawlDb, lock, fs);
+ NutchJob.cleanupAfterFailure(tempCrawlDb, lock, fs);
// throw exception so that calling routine can exit with error
throw new RuntimeException(message);
}
@@ -463,19 +464,7 @@ public void inject(Path crawlDb, Path urlDir, boolean
overwrite,
}
} catch (IOException | InterruptedException | ClassNotFoundException e) {
LOG.error("Injector job failed", e);
- cleanupAfterFailure(tempCrawlDb, lock, fs);
- throw e;
- }
- }
-
- public void cleanupAfterFailure(Path tempCrawlDb, Path lock, FileSystem fs)
- throws IOException {
- try {
- if (fs.exists(tempCrawlDb)) {
- fs.delete(tempCrawlDb, true);
- }
- LockUtil.removeLockFile(fs, lock);
- } catch (IOException e) {
+ NutchJob.cleanupAfterFailure(tempCrawlDb, lock, fs);
throw e;
}
}
diff --git a/src/java/org/apache/nutch/crawl/LinkDb.java
b/src/java/org/apache/nutch/crawl/LinkDb.java
index c6a32ba86..37bfb7bfe 100644
--- a/src/java/org/apache/nutch/crawl/LinkDb.java
+++ b/src/java/org/apache/nutch/crawl/LinkDb.java
@@ -228,8 +228,17 @@ public void invert(Path linkDb, Path[] segments, boolean
normalize,
ParseData.DIR_NAME));
}
try {
- int complete = job.waitForCompletion(true)?0:1;
- } catch (IOException | InterruptedException | ClassNotFoundException e) {
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "LinkDb job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ LockUtil.removeLockFile(fs, lock);
+ throw new RuntimeException(message);
+ }
+ } catch (IOException | InterruptedException | ClassNotFoundException e) {
+ LOG.error("LinkDb job failed {}", e);
LockUtil.removeLockFile(fs, lock);
throw e;
}
@@ -244,10 +253,18 @@ public void invert(Path linkDb, Path[] segments, boolean
normalize,
FileInputFormat.addInputPath(job, currentLinkDb);
FileInputFormat.addInputPath(job, newLinkDb);
try {
- int complete = job.waitForCompletion(true)?0:1;
- } catch (IOException e) {
- LockUtil.removeLockFile(fs, lock);
- fs.delete(newLinkDb, true);
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "LinkDb job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ NutchJob.cleanupAfterFailure(newLinkDb, lock, fs);
+ throw new RuntimeException(message);
+ }
+ } catch (IOException | InterruptedException | ClassNotFoundException e) {
+ LOG.error("LinkDb job failed {}", e);
+ NutchJob.cleanupAfterFailure(newLinkDb, lock, fs);
throw e;
}
fs.delete(newLinkDb, true);
diff --git a/src/java/org/apache/nutch/crawl/LinkDbMerger.java
b/src/java/org/apache/nutch/crawl/LinkDbMerger.java
index c8e3943b2..f2f0892a6 100644
--- a/src/java/org/apache/nutch/crawl/LinkDbMerger.java
+++ b/src/java/org/apache/nutch/crawl/LinkDbMerger.java
@@ -121,7 +121,20 @@ public void merge(Path output, Path[] dbs, boolean
normalize, boolean filter)
for (int i = 0; i < dbs.length; i++) {
FileInputFormat.addInputPath(job, new Path(dbs[i], LinkDb.CURRENT_NAME));
}
- int complete = job.waitForCompletion(true)?0:1;
+
+ try {
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "LinkDbMerge job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ throw new RuntimeException(message);
+ }
+ } catch (IOException | InterruptedException | ClassNotFoundException e) {
+ LOG.error("LinkDbMerge job failed {}", e);
+ throw e;
+ }
FileSystem fs = output.getFileSystem(getConf());
fs.mkdirs(output);
fs.rename(FileOutputFormat.getOutputPath(job), new Path(output,
diff --git a/src/java/org/apache/nutch/crawl/LinkDbReader.java
b/src/java/org/apache/nutch/crawl/LinkDbReader.java
index 8efaf0ada..519fa5996 100644
--- a/src/java/org/apache/nutch/crawl/LinkDbReader.java
+++ b/src/java/org/apache/nutch/crawl/LinkDbReader.java
@@ -166,8 +166,15 @@ public void processDumpJob(String linkdb, String output,
String regex)
job.setOutputValueClass(Inlinks.class);
try{
- int complete = job.waitForCompletion(true)?0:1;
- } catch (InterruptedException | ClassNotFoundException e){
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "LinkDbRead job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ throw new RuntimeException(message);
+ }
+ } catch (IOException | InterruptedException | ClassNotFoundException e){
LOG.error(StringUtils.stringifyException(e));
throw e;
}
diff --git a/src/java/org/apache/nutch/fetcher/Fetcher.java
b/src/java/org/apache/nutch/fetcher/Fetcher.java
index 34fb136d2..ba34d68a8 100644
--- a/src/java/org/apache/nutch/fetcher/Fetcher.java
+++ b/src/java/org/apache/nutch/fetcher/Fetcher.java
@@ -495,7 +495,14 @@ public void fetch(Path segment, int threads) throws
IOException,
job.setOutputValueClass(NutchWritable.class);
try {
- int complete = job.waitForCompletion(true)?0:1;
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "Fetcher job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ throw new RuntimeException(message);
+ }
} catch (InterruptedException | ClassNotFoundException e) {
LOG.error(StringUtils.stringifyException(e));
throw e;
diff --git a/src/java/org/apache/nutch/hostdb/ReadHostDb.java
b/src/java/org/apache/nutch/hostdb/ReadHostDb.java
index e53e0c3ed..408e3eab3 100644
--- a/src/java/org/apache/nutch/hostdb/ReadHostDb.java
+++ b/src/java/org/apache/nutch/hostdb/ReadHostDb.java
@@ -208,7 +208,7 @@ private void readHostDb(Path hostDb, Path output, boolean
dumpHomepages, boolean
throw new RuntimeException(message);
}
} catch (IOException | InterruptedException | ClassNotFoundException e) {
- LOG.error("ReadHostDb job failed", e);
+ LOG.error("ReadHostDb job failed {}", e);
throw e;
}
diff --git a/src/java/org/apache/nutch/hostdb/UpdateHostDb.java
b/src/java/org/apache/nutch/hostdb/UpdateHostDb.java
index 9119c3575..720927877 100644
--- a/src/java/org/apache/nutch/hostdb/UpdateHostDb.java
+++ b/src/java/org/apache/nutch/hostdb/UpdateHostDb.java
@@ -129,17 +129,23 @@ private void updateHostDb(Path hostDb, Path crawlDb, Path
topHosts,
conf.setClassLoader(Thread.currentThread().getContextClassLoader());
try {
- int complete = job.waitForCompletion(true)?0:1;
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "UpdateHostDb job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ NutchJob.cleanupAfterFailure(tempHostDb, lock, fs);
+ throw new RuntimeException(message);
+ }
FSUtils.replace(fs, old, current, true);
FSUtils.replace(fs, current, tempHostDb, true);
if (!preserveBackup && fs.exists(old)) fs.delete(old, true);
} catch (Exception e) {
- if (fs.exists(tempHostDb)) {
- fs.delete(tempHostDb, true);
- }
- LockUtil.removeLockFile(fs, lock);
+ LOG.error("UpdateHostDb job failed {}", e);
+ NutchJob.cleanupAfterFailure(tempHostDb, lock, fs);
throw e;
}
diff --git a/src/java/org/apache/nutch/indexer/CleaningJob.java
b/src/java/org/apache/nutch/indexer/CleaningJob.java
index a8ac64044..3ac8b9e39 100644
--- a/src/java/org/apache/nutch/indexer/CleaningJob.java
+++ b/src/java/org/apache/nutch/indexer/CleaningJob.java
@@ -162,7 +162,14 @@ public void delete(String crawldb, boolean noCommit)
conf.setBoolean(IndexerMapReduce.INDEXER_DELETE, true);
try{
- int complete = job.waitForCompletion(true)?0:1;
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "CleaningJob did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ throw new RuntimeException(message);
+ }
} catch (InterruptedException | ClassNotFoundException e) {
LOG.error(StringUtils.stringifyException(e));
throw e;
diff --git a/src/java/org/apache/nutch/indexer/IndexingJob.java
b/src/java/org/apache/nutch/indexer/IndexingJob.java
index e4997cb89..67572915b 100644
--- a/src/java/org/apache/nutch/indexer/IndexingJob.java
+++ b/src/java/org/apache/nutch/indexer/IndexingJob.java
@@ -146,8 +146,15 @@ public void index(Path crawlDb, Path linkDb, List<Path>
segments,
FileOutputFormat.setOutputPath(job, tmp);
try {
try{
- int complete = job.waitForCompletion(true)?0:1;
- } catch (InterruptedException | ClassNotFoundException e) {
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "Indexing job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ throw new RuntimeException(message);
+ }
+ } catch (IOException | InterruptedException | ClassNotFoundException e) {
LOG.error(StringUtils.stringifyException(e));
throw e;
}
diff --git a/src/java/org/apache/nutch/parse/ParseSegment.java
b/src/java/org/apache/nutch/parse/ParseSegment.java
index 61aa99740..1d6446325 100644
--- a/src/java/org/apache/nutch/parse/ParseSegment.java
+++ b/src/java/org/apache/nutch/parse/ParseSegment.java
@@ -259,8 +259,15 @@ public void parse(Path segment) throws IOException,
job.setOutputValueClass(ParseImpl.class);
try{
- int complete = job.waitForCompletion(true)?0:1;
- } catch (InterruptedException | ClassNotFoundException e) {
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "Parse job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ throw new RuntimeException(message);
+ }
+ } catch (IOException | InterruptedException | ClassNotFoundException e) {
LOG.error(StringUtils.stringifyException(e));
throw e;
}
diff --git a/src/java/org/apache/nutch/segment/SegmentMerger.java
b/src/java/org/apache/nutch/segment/SegmentMerger.java
index 780e10a6f..188ae69d2 100644
--- a/src/java/org/apache/nutch/segment/SegmentMerger.java
+++ b/src/java/org/apache/nutch/segment/SegmentMerger.java
@@ -738,7 +738,19 @@ public void merge(Path out, Path[] segs, boolean filter,
boolean normalize,
setConf(conf);
- int complete = job.waitForCompletion(true)?0:1;
+ try {
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "SegmentMerger job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ throw new RuntimeException(message);
+ }
+ } catch (IOException | InterruptedException | ClassNotFoundException e) {
+ LOG.error("SegmentMerger job failed {}", e);
+ throw e;
+ }
}
/**
diff --git a/src/java/org/apache/nutch/segment/SegmentReader.java
b/src/java/org/apache/nutch/segment/SegmentReader.java
index 28b88cd6e..0b65a2b81 100644
--- a/src/java/org/apache/nutch/segment/SegmentReader.java
+++ b/src/java/org/apache/nutch/segment/SegmentReader.java
@@ -244,7 +244,14 @@ public void dump(Path segment, Path output) throws
IOException,
job.setOutputValueClass(NutchWritable.class);
try {
- int complete = job.waitForCompletion(true)?0:1;
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "SegmentReader job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ throw new RuntimeException(message);
+ }
} catch (IOException | InterruptedException | ClassNotFoundException e ){
LOG.error(StringUtils.stringifyException(e));
throw e;
diff --git a/src/java/org/apache/nutch/tools/FreeGenerator.java
b/src/java/org/apache/nutch/tools/FreeGenerator.java
index 3b01bb431..ab5109edf 100644
--- a/src/java/org/apache/nutch/tools/FreeGenerator.java
+++ b/src/java/org/apache/nutch/tools/FreeGenerator.java
@@ -201,8 +201,15 @@ public int run(String[] args) throws Exception {
FileOutputFormat.setOutputPath(job, new Path(args[1], new Path(segName,
CrawlDatum.GENERATE_DIR_NAME)));
try {
- int complete = job.waitForCompletion(true)?0:1;
- } catch (Exception e) {
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "FreeGenerator job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ throw new RuntimeException(message);
+ }
+ } catch (IOException | InterruptedException | ClassNotFoundException e) {
LOG.error("FAILED: " + StringUtils.stringifyException(e));
return -1;
}
diff --git a/src/java/org/apache/nutch/tools/arc/ArcSegmentCreator.java
b/src/java/org/apache/nutch/tools/arc/ArcSegmentCreator.java
index 1f9e660f6..499b24607 100644
--- a/src/java/org/apache/nutch/tools/arc/ArcSegmentCreator.java
+++ b/src/java/org/apache/nutch/tools/arc/ArcSegmentCreator.java
@@ -395,7 +395,14 @@ public void createSegments(Path arcFiles, Path
segmentsOutDir)
job.setOutputValueClass(NutchWritable.class);
try {
- int complete = job.waitForCompletion(true)?0:1;
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "ArcSegmentCreator job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ throw new RuntimeException(message);
+ }
} catch (IOException | InterruptedException | ClassNotFoundException e){
LOG.error(StringUtils.stringifyException(e));
throw e;
diff --git a/src/java/org/apache/nutch/tools/warc/WARCExporter.java
b/src/java/org/apache/nutch/tools/warc/WARCExporter.java
index aae8064f8..2921a9763 100644
--- a/src/java/org/apache/nutch/tools/warc/WARCExporter.java
+++ b/src/java/org/apache/nutch/tools/warc/WARCExporter.java
@@ -285,13 +285,20 @@ public int generateWARC(String output, List<Path>
segments) throws IOException{
job.setOutputValueClass(WARCWritable.class);
try {
- int complete = job.waitForCompletion(true)?0:1;
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ String message = "WARCExporter job did not succeed, job status:"
+ + job.getStatus().getState() + ", reason: "
+ + job.getStatus().getFailureInfo();
+ LOG.error(message);
+ throw new RuntimeException(message);
+ }
LOG.info(job.getCounters().toString());
long end = System.currentTimeMillis();
LOG.info("WARCExporter: finished at {}, elapsed: {}", sdf.format(end),
TimingUtil.elapsedTime(start, end));
- } catch (Exception e) {
- LOG.error("Exception caught", e);
+ } catch (IOException | InterruptedException | ClassNotFoundException e) {
+ LOG.error("WARCExporter job failed {}", e);
return -1;
}
diff --git a/src/java/org/apache/nutch/util/NutchJob.java
b/src/java/org/apache/nutch/util/NutchJob.java
index 34a9acdb6..06f1cc2c5 100644
--- a/src/java/org/apache/nutch/util/NutchJob.java
+++ b/src/java/org/apache/nutch/util/NutchJob.java
@@ -19,6 +19,8 @@
import java.io.IOException;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.Job;
@@ -34,4 +36,19 @@ public static Job getInstance(Configuration conf) throws
IOException {
return Job.getInstance(conf);
}
+ /*
+ * Clean up the file system in case of a job failure.
+ */
+ public static void cleanupAfterFailure(Path tempDir, Path lock, FileSystem
fs)
+ throws IOException {
+ try {
+ if (fs.exists(tempDir)) {
+ fs.delete(tempDir, true);
+ }
+ LockUtil.removeLockFile(fs, lock);
+ } catch (IOException e) {
+ throw e;
+ }
+ }
+
}
diff --git a/src/java/org/apache/nutch/util/SitemapProcessor.java
b/src/java/org/apache/nutch/util/SitemapProcessor.java
index 380ac0710..70f43727f 100644
--- a/src/java/org/apache/nutch/util/SitemapProcessor.java
+++ b/src/java/org/apache/nutch/util/SitemapProcessor.java
@@ -52,6 +52,7 @@
import org.apache.nutch.protocol.ProtocolFactory;
import org.apache.nutch.protocol.ProtocolOutput;
import org.apache.nutch.protocol.ProtocolStatus;
+import org.apache.nutch.util.NutchJob;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -383,7 +384,7 @@ public void sitemap(Path crawldb, Path hostdb, Path
sitemapUrlDir, boolean stric
+ " job did not succeed, job status: " + job.getStatus().getState()
+ ", reason: " + job.getStatus().getFailureInfo();
LOG.error(message);
- cleanupAfterFailure(tempCrawlDb, lock, fs);
+ NutchJob.cleanupAfterFailure(tempCrawlDb, lock, fs);
// throw exception so that calling routine can exit with error
throw new RuntimeException(message);
}
@@ -415,19 +416,7 @@ public void sitemap(Path crawldb, Path hostdb, Path
sitemapUrlDir, boolean stric
}
} catch (IOException | InterruptedException | ClassNotFoundException e) {
LOG.error("SitemapProcessor_" + crawldb.toString(), e);
- cleanupAfterFailure(tempCrawlDb, lock, fs);
- throw e;
- }
- }
-
- public void cleanupAfterFailure(Path tempCrawlDb, Path lock, FileSystem fs)
- throws IOException {
- try {
- if (fs.exists(tempCrawlDb)) {
- fs.delete(tempCrawlDb, true);
- }
- LockUtil.removeLockFile(fs, lock);
- } catch (IOException e) {
+ NutchJob.cleanupAfterFailure(tempCrawlDb, lock, fs);
throw e;
}
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Must check return value of job.waitForCompletion()
> --------------------------------------------------
>
> Key: NUTCH-2518
> URL: https://issues.apache.org/jira/browse/NUTCH-2518
> Project: Nutch
> Issue Type: Bug
> Components: crawldb, fetcher, generator, hostdb, linkdb
> Affects Versions: 1.15
> Reporter: Sebastian Nagel
> Priority: Blocker
> Fix For: 1.15
>
>
> The return value of job.waitForCompletion() of the new MapReduce API
> (NUTCH-2375) must always be checked. If it's not true, the job has been
> failed or killed. Accordingly, the program
> - should not proceed with further jobs/steps
> - must clean-up temporary data, unlock CrawlDB, etc.
> - exit with non-zero exit value, so that scripts running the crawl workflow
> can handle the failure
> Cf. NUTCH-2076, NUTCH-2442, [NUTCH-2375 PR
> #221|https://github.com/apache/nutch/pull/221#issuecomment-332941883].
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
