Kafka setting consumer offset before committed offset

user_440237 0
Viewed 808

I'm running a program that starts with a message in the topic, consumes it, processes it, commits the next offset, and publishes a new message to the same topic, all in a transactional fashion. I have the following (simplified) trace:

Fetch READ_COMMITTED at offset 20 for partition test-topic-0
processing message at offset 20
Committed offset 21 for partition test-topic-0
Sending PRODUCE
COMMITTING_TRANSACTION
Fetch READ_COMMITTED at offset 22 for partition test-topic-0
processing message at offset 22 <==== first time
...rebalance...
Setting offset for partition test-topic-0 to the committed offset FetchPosition{offset=21
Committed offset 23 for partition test-topic-0
Sending PRODUCE
COMMITTING_TRANSACTION
Fetch READ_COMMITTED at offset 24 for partition test-topic-0
stale fetch response for partition test-topic-0 since its offset 24 does not match the expected offset FetchPosition{offset=21
Fetch READ_COMMITTED at offset 21 for partition test-topic-0
processing message at offset 22 <==== second time

As a result of this I process the message "22" twice. Is it expected for kafka to just rewind the consumer offset to before the committed offset? Does the ordering of the log look right? I can update the question with the full log if necessary but I don't think there is anything useful there.

kafka-transactions-api kafka-producer-api kafka-consumer-api apache-kafka message
1 Answers

Looks like a rebalance occurred before the producer could complete the transaction. Would be helpful to see the code / configs you're using / version of Kafka.

The transactional consume-process-produce requires the producer to do a couple of different things. When processing a batch of records:

  • producer.beginTransaction() - this method guarantees that everything produced from time it was called until the transaction is aborted/committed, to be part of a single transaction.

  • producer.send(producerRecord) - for each message you process in the batch.

  • producer.sendOffsetsToTransaction( Map<TopicPartition, OffsetAndMetadata> offsetsToCommit, consumer.groupMetadata() ) - once you've gone through the batch, which will commit the offsets as part of the transaction. Note that committing offsets any other way will not provide transactional guarantees.

Once all records from the batch have been produced and you committed offsets as part of the transaction, you finally commit the transaction and seal the deal - producer.commitTransaction()

With that said, this should explain why it rejected message 24 and reprocessed message 22. I believe message 23 did not get to the last producer step, but would need to see the code to be sure. From the Kafka definitive guide:

In order to guarantee that messages will be read in order, read_committed mode will not return messages that were produced after the point when the first still-open transaction began (known as the Last Stable Offset, or LSO). Those messages will be witheld until that transaction is committed or aborted by the producer, or until the reach transaction.timeout.ms (15 min default) and are aborted by the broker.

And

The two main mistakes (to transactions) are assuming that exactly once guarantees apply on actions other than producing to Kafka, and that consumers always read entire transactions and have information about transaction boundaries.

user_144750340
Related