[GitHub] [kafka] guozhangwang commented on a change in pull request #11151: MINOR: Should commit a task if the consumer position advanced as well
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 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
[GitHub] [kafka] guozhangwang commented on a change in pull request #11151: MINOR: Should commit a task if the consumer position advanced as well
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 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
