[KAFKA-6607] Kafka Streams lag not zero when input topic transactional

• Type: Improvement
• Status: Resolved
• Priority: Minor
• Resolution: Fixed
• Affects Version/s: 1.0.0
• Fix Version/s: 2.5.0
• Component/s: streams
• Assignee: Matthias J. Sax
• Reporter: Matthias J. Sax
• Created: 02/Mar/18 23:50
• Updated: 06/Dec/22 00:27
• Resolved: 12/Feb/20 20:25

Description

When an input topic for a Kafka Streams application is written using transaction, Kafka Streams commits an "incorrect" offset, ie, it commits "lastProcessedMessageOffset + 1" that is smaller than "endOffset" if it reaches the end of topic. The reason is the commit marker that is the last "message" in the topic; Streams does not take commit markers into account when committing.

This is not a correctness issue, but when one inspect the consumer lag via bin/kafka-consumer.group.sh the lag is shown as 1 instead of 0 – what is correct from consumer-group tool point of view.

Note that all applications using a plain consumer may face the same issue if they use `KafkaConsumer#commitSync(Map<TopicPartition, OffsetAndMetadata> offsets)`: to address the issue, the correct pattern is to either commit "nextRecord.offset()" (if the next record is available already, ie, was returned by `poll()`, or use `consumer.position()` that takes the commit marker into account and would "step over it").

Issue Links

• is related to: KAFKA-13404 Kafka sink connectors do not commit offset correctly if messages are produced in transaction
• is related to: KAFKA-12980 Allow consumers to return from poll when position advances due to aborted transactions
• relates to: KAFKA-10683 Consumer.position() Ignores Transaction Marker with read_uncommitted
• relates to: KAFKA-12951 Infinite loop while restoring a GlobalKTable
• relates to: KAFKA-14442 GlobalKTable restoration waits requestTimeout during application restart
• links to: GitHub Pull Request #8040
• links to: GitHub Pull Request #8091