commit c784f4da46d1cf594aa4156135b9c196aa66d931
author chaitanya <chaithu@apache.org> Thu May 18 16:07:52 2017 +0530
committer chaitanya <chaithu@apache.org> Thu May 18 16:07:52 2017 +0530
tree dad8ab358a2cd5d61de091fc23fd568205a795f6
parent 3a362984143dde0a89fb9538a704ed1f444f0204

APEXMALHAR-2493 Fixed the issue of KafkaSinglePortExactlyOnceOutputOperator going to the blocked state during recovery

kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaSinglePortExactlyOnceOutputOperator.java

diff --git a/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaSinglePortExactlyOnceOutputOperator.java b/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaSinglePortExactlyOnceOutputOperator.java
index 75af448..23c519f 100644
--- a/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaSinglePortExactlyOnceOutputOperator.java
+++ b/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaSinglePortExactlyOnceOutputOperator.java
@@ -352,6 +352,11 @@
 
         for (ConsumerRecord<String, T> consumerRecord : consumerRecords) {
 
+          if (consumerRecord.offset() >= currentOffset) {
+            crossedBoundary = true;
+            break;
+          }
+
           if (!doesKeyBelongsToThisInstance(operatorId, consumerRecord.key())) {
             continue;
           }
@@ -365,10 +370,6 @@
             partialWindowTuples.put(value, 1);
           }
 
-          if (consumerRecord.offset() >= currentOffset) {
-            crossedBoundary = true;
-            break;
-          }
         }
 
         if (crossedBoundary) {