Issue dataArtisans#61 - FlinkFlowStepStats.getCountersFor should retu…

…rn counter names
kkrugler · May 9, 2016 · 1ae8653 · 1ae8653
1 parent 70341cb
commit 1ae8653
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 5 deletions.
diff --git a/src/main/java/com/dataartisans/flink/cascading/planner/FlinkFlowStepStats.java b/src/main/java/com/dataartisans/flink/cascading/planner/FlinkFlowStepStats.java
@@ -81,15 +81,16 @@ public Collection<String> getCountersFor(String group) {
 
 		for (String key : currentAccumulators.keySet()) {
 			if (EnumStringConverter.accInGroup(group, key)) {
-				result.add(key);
+				result.add(EnumStringConverter.groupCounterToCounter(key));
 			}
 		}
+
 		return result;
 	}
 
 	@Override
 	public long getCounterValue(Enum counter) {
-		return getCounterValue(EnumStringConverter.enumToGroup(counter), EnumStringConverter.enumToKey(counter));
+		return getCounterValue(EnumStringConverter.enumToGroup(counter), EnumStringConverter.enumToCounter(counter));
 	}
 
 	@Override

diff --git a/src/main/java/com/dataartisans/flink/cascading/runtime/stats/EnumStringConverter.java b/src/main/java/com/dataartisans/flink/cascading/runtime/stats/EnumStringConverter.java
@@ -22,7 +22,13 @@ public static String enumToGroup(Enum e) {
 		return e.getDeclaringClass().getName();
 	}
 
-	public static String enumToKey(Enum e) {
+	/**
+	 * Convert an Enum to the counter portion of its name.
+	 * 
+	 * @param e
+	 * @return
+	 */
+	public static String enumToCounter(Enum e) {
 		return e.name();
 	}
 
@@ -42,4 +48,8 @@ public static String groupCounterToGroup(String groupKey) {
 		return groupKey.split("->")[0];
 	}
 
+	public static String groupCounterToCounter(String groupKey) {
+		return groupKey.split("->")[1];
+	}
+
 }
diff --git a/src/main/java/com/dataartisans/flink/cascading/runtime/util/FlinkFlowProcess.java b/src/main/java/com/dataartisans/flink/cascading/runtime/util/FlinkFlowProcess.java
@@ -119,7 +119,7 @@ public void keepAlive() {
 
 	@Override
 	public void increment(Enum e, long l) {
-		increment(EnumStringConverter.enumToGroup(e), EnumStringConverter.enumToKey(e), l);
+		increment(EnumStringConverter.enumToGroup(e), EnumStringConverter.enumToCounter(e), l);
 	}
 
 	@Override
@@ -132,7 +132,7 @@ public void increment(String group, String counter, long l) {
 
 	@Override
 	public long getCounterValue(Enum e) {
-		return getCounterValue(EnumStringConverter.enumToGroup(e), EnumStringConverter.enumToKey(e));
+		return getCounterValue(EnumStringConverter.enumToGroup(e), EnumStringConverter.enumToCounter(e));
 	}
 
 	@Override