tez-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ss...@apache.org
Subject tez git commit: TEZ-3582. Exception swallowed in PipelinedSorter causing incorrect results (rbalamohan)
Date Thu, 26 Jan 2017 00:29:35 GMT
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 <rbalamohan@apache.org>
Authored: Wed Jan 25 13:02:53 2017 +0530
Committer: Siddharth Seth <sseth@apache.org>
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);
       }
     }
 


Mime
View raw message