[FLINK-2382] fix Flink live accumulators for TwoInputStreamTask

MoeweX · Jul 21, 2015 · 1d373a7 · 1d373a7
1 parent 0f589aa
commit 1d373a7
Showing 1 changed file with 6 additions and 1 deletion.
diff --git a/...aming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTask.java b/...aming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTask.java
@@ -22,6 +22,7 @@
 import java.util.List;
 
 import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
 import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
 import org.apache.flink.streaming.api.graph.StreamEdge;
 import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
@@ -66,6 +67,10 @@ public void registerInputOutput() {
 
 		inputProcessor = new StreamTwoInputProcessor<IN1, IN2>(inputList1, inputList2, inputDeserializer1, inputDeserializer2, getExecutionConfig().areTimestampsEnabled());
 
+		AccumulatorRegistry registry = getEnvironment().getAccumulatorRegistry();
+		AccumulatorRegistry.Reporter reporter = registry.getReadWriteReporter();
+		inputProcessor.setReporter(reporter);
+
 		inputProcessor.registerTaskEventListener(getCheckpointBarrierListener(), CheckpointBarrier.class);
 	}
 
@@ -107,7 +112,7 @@ public void invoke() throws Exception {
 					LOG.warn("Exception while closing operator.", t);
 				}
 			}
-			
+
 			throw e;
 		}
 		finally {