TEZ-1808. Job can fail since name of intermediate files can be too long
in specific situation. Contributed by Tsuyoshi OZAWA.
(cherry picked from commit 87b4c6b40cbe607b9d16b9d7e4a8f965757b5196)
Conflicts:
CHANGES.txt
(cherry picked from commit 20c04e5534cddc20702fd025c407a2ccadc39fe9)
diff --git a/CHANGES.txt b/CHANGES.txt
index 945c645..5459112 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -18,6 +18,7 @@
TEZ-1796. Use of DeprecationDelta broke build against 2.2 Hadoop.
TEZ-1818. Problem loading tez-api-version-info.properties in case current context classloader
in not pointing to Tez jars.
+ TEZ-1808. Job can fail since name of intermediate files can be too long in specific situation.
Release 0.5.2: 2014-11-07
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/MergeManager.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/MergeManager.java
index 6081f91..fd8b1ea 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/MergeManager.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/MergeManager.java
@@ -26,6 +26,8 @@
import java.util.TreeSet;
import com.google.common.annotations.VisibleForTesting;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.io.FilenameUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
@@ -130,6 +132,7 @@
private final int ifileReadAheadLength;
private final int ifileBufferSize;
+ private AtomicInteger mergeFileSequenceId = new AtomicInteger(0);
/**
* Construct the MergeManager. Must call start before it becomes usable.
@@ -696,10 +699,13 @@
} else {
namePart = file0.getPath().getName().toString();
}
- Path outputPath = localDirAllocator.getLocalPathForWrite(namePart, approxOutputSize, conf);
- outputPath = outputPath.suffix(Constants.MERGED_OUTPUT_PREFIX);
- Writer writer =
+ // namePart includes the suffix of the file. We need to remove it.
+ namePart = FilenameUtils.removeExtension(namePart);
+ Path outputPath = localDirAllocator.getLocalPathForWrite(namePart, approxOutputSize, conf);
+ outputPath = outputPath.suffix(Constants.MERGED_OUTPUT_PREFIX + mergeFileSequenceId.getAndIncrement());
+
+ Writer writer =
new Writer(conf, rfs, outputPath,
(Class)ConfigUtils.getIntermediateInputKeyClass(conf),
(Class)ConfigUtils.getIntermediateInputValueClass(conf),
