Repository: tez Updated Branches: refs/heads/branch-0.8 567e23132 -> c797d6e3f
TEZ-3582. Exception swallowed in PipelinedSorter causing incorrect results (rbalamohan) (cherry picked from commit abab526940f6353866d866b93d6da685edfa6014) Project: http://git-wip-us.apache.org/repos/asf/tez/repo Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/c797d6e3 Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/c797d6e3 Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/c797d6e3 Branch: refs/heads/branch-0.8 Commit: c797d6e3f34616b4dfc5ac17ca32cbb71cfebca9 Parents: 567e231 Author: Rajesh Balamohan <[email protected]> Authored: Wed Jan 25 13:02:53 2017 +0530 Committer: Siddharth Seth <[email protected]> Committed: Wed Jan 25 15:50:17 2017 -0800 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../common/sort/impl/PipelinedSorter.java | 20 ++++++++++++++++---- 2 files changed, 17 insertions(+), 4 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/tez/blob/c797d6e3/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index f3613ec..6c1e7c1 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -7,6 +7,7 @@ INCOMPATIBLE CHANGES ALL CHANGES: + TEZ-3582. Exception swallowed in PipelinedSorter causing incorrect results. TEZ-3462. Task attempt failure during container shutdown loses useful container diagnostics TEZ-3574. Container reuse won't pickup extra dag level local resource. TEZ-3566. Avoid caching fs isntances in TokenCache after a point. http://git-wip-us.apache.org/repos/asf/tez/blob/c797d6e3/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/PipelinedSorter.java ---------------------------------------------------------------------- diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/PipelinedSorter.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/PipelinedSorter.java index 5695bde..8ccc3c2 100644 --- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/PipelinedSorter.java +++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/PipelinedSorter.java @@ -335,6 +335,7 @@ public class PipelinedSorter extends ExternalSorter { } else { // queue up the sort SortTask task = new SortTask(span, sorter); + LOG.debug("Submitting span={} for sort", span.toString()); Future<SpanIterator> future = sortmaster.submit(task); merger.add(future); span = newSpan; @@ -969,8 +970,15 @@ public class PipelinedSorter extends ExternalSorter { items = 1024*1024; perItem = 16; } - newSpan = new SortSpan(remaining, items, perItem, - ConfigUtils.getIntermediateOutputKeyComparator(conf)); + final RawComparator newComparator = ConfigUtils.getIntermediateOutputKeyComparator(conf); + if (this.comparator == newComparator) { + LOG.warn("Same comparator used. comparator={}, newComparator={}," + + " hashCode: comparator={}, newComparator={}", + this.comparator, newComparator, + System.identityHashCode(this.comparator), + System.identityHashCode(newComparator)); + } + newSpan = new SortSpan(remaining, items, perItem, newComparator); newSpan.index = index+1; LOG.info(String.format(outputContext.getDestinationVertexName() + ": " + "New Span%d.length = %d, perItem = %d", newSpan.index, newSpan .length(), perItem) + ", counter:" + mapOutputRecordCounter.getValue()); @@ -1278,6 +1286,7 @@ public class PipelinedSorter extends ExternalSorter { } public final boolean ready() throws IOException, InterruptedException { + int numSpanItr = futures.size(); try { SpanIterator iter = null; while(this.futures.size() > 0) { @@ -1299,8 +1308,11 @@ public class PipelinedSorter extends ExternalSorter { LOG.info(outputContext.getDestinationVertexName() + ": " + "Heap = " + sb.toString()); return true; } catch(ExecutionException e) { - LOG.info(outputContext.getDestinationVertexName() + ": " + e.toString()); - return false; + LOG.error("Heap size={}, total={}, eq={}, partition={}, gallop={}, totalItr={}," + + " futures.size={}, destVertexName={}", + heap.size(), total, eq, partition, gallop, numSpanItr, futures.size(), + outputContext.getDestinationVertexName(), e); + throw new IOException(e); } }
