[GitHub] [kafka] vvcephei commented on a change in pull request #11151: MINOR: Should commit a task if the consumer position advanced as well

[GitBox <gi...@apache.org>]

vvcephei commented on a change in pull request #11151:
URL: https://github.com/apache/kafka/pull/11151#discussion_r680997728



##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -1158,7 +1158,38 @@ public String toString(final String indent) {
 
     @Override
     public boolean commitNeeded() {
-        return commitNeeded;
+        // we need to do an extra check if the flag was false, that
+        // if the consumer position has been updated; this is because
+        // there may be non data records such as control markers bypassed
+        if (commitNeeded) {
+            return true;
+        } else {
+            for (final Map.Entry<TopicPartition, Long> entry : consumedOffsets.entrySet()) {
+                final TopicPartition partition = entry.getKey();
+                try {
+                    final long offset = mainConsumer.position(partition);
+
+                    // note the position in consumer is the "next" record to fetch,
+                    // so it should be larger than the consumed offset by 1; if it is
+                    // more than 1 it means there are skipped offsets

Review comment:
       "skipped offsets" sounds scary. How about "control records, which the consumer skips over silently"?




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org