guozhangwang commented on a change in pull request #11151: URL: https://github.com/apache/kafka/pull/11151#discussion_r682935981
########## 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: ack -- 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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org