[FLINK-4833] properly log exceptions in CountMinHeavyHitter

This closes apache#2660
naraee · Oct 21, 2016 · a27451d · a27451d
1 parent 0d29035
commit a27451d
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 2 deletions.
diff --git a/...in/java/org/apache/flink/contrib/operatorstatistics/heavyhitters/CountMinHeavyHitter.java b/...in/java/org/apache/flink/contrib/operatorstatistics/heavyhitters/CountMinHeavyHitter.java
@@ -123,9 +123,9 @@ public void merge(HeavyHitter toMerge) throws CMHeavyHitterMergeException {
 			cardinality+=cmToMerge.cardinality;
 
 		}catch (ClassCastException ex){
-			throw new CMHeavyHitterMergeException("Both heavy hitter objects must belong to the same class");
+			throw new CMHeavyHitterMergeException("Both heavy hitter objects must belong to the same class", ex);
 		}catch (Exception ex){
-			throw new CMHeavyHitterMergeException("Cannot merge count min sketches: "+ex.getMessage());
+			throw new CMHeavyHitterMergeException("Cannot merge count min sketches: ", ex);
 		}
 	}
 
@@ -138,9 +138,14 @@ public HashMap<Object,Long> getHeavyHitters() {
 	}
 
 	protected static class CMHeavyHitterMergeException extends HeavyHitterMergeException {
+
 		public CMHeavyHitterMergeException(String message) {
 			super(message);
 		}
+
+		public CMHeavyHitterMergeException(String message, Throwable cause) {
+			super(message, cause);
+		}
 	}
 
 	@Override

diff --git a/...a/org/apache/flink/contrib/operatorstatistics/heavyhitters/HeavyHitterMergeException.java b/...a/org/apache/flink/contrib/operatorstatistics/heavyhitters/HeavyHitterMergeException.java
@@ -25,4 +25,8 @@ public class HeavyHitterMergeException extends Exception {
 	public HeavyHitterMergeException(String message) {
 		super(message);
 	}
+
+	public HeavyHitterMergeException(String message, Throwable cause) {
+		super(message, cause);
+	}
 }