EndOffset and committed offset in transaction

[Ludwig Schmid <ls...@gmail.com>]

Hello,

in an application I use a producer and a consumer. The consumer polls data from the same topic (replication-factor 1) the producer sends to. The producer works transactional and within a transaction
the producer sends records to the topic and also sends offsets. The application looks something like this:

consumer.poll(Duration)
// Do some processing with polled records
producer.beginTransaction();
producer.send(new ProducerRecord<>(...));
producer.sendOffsetsToTransaction(offsets, GROUP_ID);
producer.commitTransaction();

Call to consumer.committed(topicPartition) and consumer.endOffset(topicPartition) immediately after the producer.commitTransaction() do not return the expected values.
For example, the produced record has offset 36, returned end offset is 31. The sendOffsetToTransactio is 28, returned committed offset is 24. However in the next loop, the commited offset returned is 28.

This looks strange to me, since, according to the docs, the transactional APIs are blocking. Therefore I would be happy, if you could explain the described behavior.

Kafka version: kafka_2.12-2.1.0

Thank you in advandce,
Ludwig

Re: EndOffset and committed offset in transaction

["Matthias J. Sax" <ma...@confluent.io>]

Did you put transaction markers into account?

Each time a transaction is committed or aborted, a commit or abort
marker is written that occupies one offset.


-Matthias


On 10/17/19 8:38 AM, Ludwig Schmid wrote:
> Hello,
> 
> in an application I use a producer and a consumer. The consumer polls data from the same topic (replication-factor 1) the producer sends to. The producer works transactional and within a transaction
> the producer sends records to the topic and also sends offsets. The application looks something like this:
> 
> consumer.poll(Duration)
> // Do some processing with polled records
> producer.beginTransaction();
> producer.send(new ProducerRecord<>(...));
> producer.sendOffsetsToTransaction(offsets, GROUP_ID);
> producer.commitTransaction();
> 
> Call to consumer.committed(topicPartition) and consumer.endOffset(topicPartition) immediately after the producer.commitTransaction() do not return the expected values.
> For example, the produced record has offset 36, returned end offset is 31. The sendOffsetToTransactio is 28, returned committed offset is 24. However in the next loop, the commited offset returned is 28.
> 
> This looks strange to me, since, according to the docs, the transactional APIs are blocking. Therefore I would be happy, if you could explain the described behavior.
> 
> Kafka version: kafka_2.12-2.1.0
> 
> Thank you in advandce,
> Ludwig
>