Updated Branches: refs/heads/master f82a4c9dc -> c0c8dbed7
Fix for mapreduce timeouts, progressable object was not being set in the ProgressableDirectory. It had to do with when the DefaultBlurReducer set the progressable object and when the directories were setup for indexing. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/c0c8dbed Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/c0c8dbed Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/c0c8dbed Branch: refs/heads/master Commit: c0c8dbed73c5f87fffebe35fb3ad86b7d930d8d3 Parents: f82a4c9 Author: Aaron McCurry <[email protected]> Authored: Mon Dec 2 20:34:20 2013 -0500 Committer: Aaron McCurry <[email protected]> Committed: Mon Dec 2 20:34:36 2013 -0500 ---------------------------------------------------------------------- .../blur/mapreduce/lib/BlurOutputFormat.java | 17 ++++++++++--- .../mapreduce/lib/ProgressableDirectory.java | 25 +++++++++++++------- 2 files changed, 31 insertions(+), 11 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/c0c8dbed/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurOutputFormat.java ---------------------------------------------------------------------- diff --git a/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurOutputFormat.java b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurOutputFormat.java index 4644761..0dbc887 100644 --- a/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurOutputFormat.java +++ b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurOutputFormat.java @@ -391,8 +391,7 @@ public class BlurOutputFormat extends OutputFormat<Text, BlurMutate> { String shardName = BlurUtil.getShardName(BlurConstants.SHARD_PREFIX, shardId); Path indexPath = new Path(tableOutput, shardName); _newIndex = new Path(indexPath, tmpDirName); - _finalDir = new ProgressableDirectory(new HdfsDirectory(configuration, _newIndex), - BlurOutputFormat.getProgressable()); + _finalDir = new ProgressableDirectory(new HdfsDirectory(configuration, _newIndex), getProgressable()); _finalDir.setLockFactory(NoLockFactory.getNoLockFactory()); TableContext tableContext = TableContext.create(tableDescriptor); @@ -410,7 +409,7 @@ public class BlurOutputFormat extends OutputFormat<Text, BlurMutate> { if (_indexLocally) { String localDirPath = System.getProperty(JAVA_IO_TMPDIR); _localPath = new File(localDirPath, UUID.randomUUID().toString() + ".tmp"); - _localDir = new ProgressableDirectory(FSDirectory.open(_localPath), BlurOutputFormat.getProgressable()); + _localDir = new ProgressableDirectory(FSDirectory.open(_localPath), getProgressable()); _writer = new IndexWriter(_localDir, _conf.clone()); } else { _localPath = null; @@ -419,6 +418,18 @@ public class BlurOutputFormat extends OutputFormat<Text, BlurMutate> { } } + private Progressable getProgressable() { + return new Progressable() { + @Override + public void progress() { + Progressable progressable = BlurOutputFormat.getProgressable(); + if (progressable != null) { + progressable.progress(); + } + } + }; + } + @Override public void write(Text key, BlurMutate value) throws IOException, InterruptedException { if (!_countersSetup) { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/c0c8dbed/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/ProgressableDirectory.java ---------------------------------------------------------------------- diff --git a/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/ProgressableDirectory.java b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/ProgressableDirectory.java index 6a96166..ef4dd78 100644 --- a/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/ProgressableDirectory.java +++ b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/ProgressableDirectory.java @@ -20,6 +20,8 @@ import java.io.IOException; import java.util.Collection; import java.util.Map; +import org.apache.blur.log.Log; +import org.apache.blur.log.LogFactory; import org.apache.hadoop.util.Progressable; import org.apache.lucene.store.BufferedIndexInput; import org.apache.lucene.store.DataInput; @@ -37,17 +39,24 @@ import org.apache.lucene.store.LockFactory; */ public class ProgressableDirectory extends Directory { - private Directory _directory; - private Progressable _progressable; + private static final Log LOG = LogFactory.getLog(ProgressableDirectory.class); + + private final Directory _directory; + private final Progressable _progressable; public ProgressableDirectory(Directory directory, Progressable progressable) { _directory = directory; - _progressable = progressable == null ? new Progressable() { - @Override - public void progress() { - - } - } : progressable; + if (progressable == null) { + LOG.warn("Progressable is null."); + _progressable = new Progressable() { + @Override + public void progress() { + + } + }; + } else { + _progressable = progressable; + } } @Override
