Page History

...

Note that for transactions with 2PC enabled the TransactionTimeoutMs would be set to Integer.MAX_VALUE. -1.

Let's consider some examples of the state transitions and how the various producer ids and epochs are used.

...

• InitProducerId(true) may be issued multiple times (e.g. client gets into a crash loop).  The ProducerId and ProducerEpoch of the ongoing transaction always stay the same, but the NextProducerEpoch is always incremented.  Eventually, NextProducerEpoch may overflow, in which case we can allocate a new NextProducerId.
• When a commit request is sent, it uses the latest ProducerId and ProducerEpoch.  We send out markers using the original ongoing transaction's ProducerId and ProducerEpoch + 1, but the next transaction will use the latest ProducerId and ProducerEpoch + 1 (this is what the response is going to contain).  It may happen (like in this example) that the latest ProducerEpoch is already at MAX, in which case we'd need to allocate a new ProducerId.  In order to support retries we store the previous ProducerId in the PrevProducerId.  Thus in such situation the PrepareCommit state can have three distinct producer ids:
  • ProducerId – this is used to send our commit markers
  • NextProducerId – this is the producer id to use for the next transaction
  • PrevProducerId – this is the producer id to avoid self-fencing on retries (i.e. if the commit request times out and the client retries with previous producer id, we can return success and new producer id)

...

If the value is 'true' then the corresponding field is set in the InitProducerIdRequest and the KafkaProducer object is set into a state which only allows calling calling  .commitTransaction, .abortTransaction, or .abortTransactioncompleteTransaction.

New method will be added to KafkaProducer:

...

This would flush all the pending messages and transition the producer into a mode where only .commitTransaction, .abortTransaction, or .completeTransaction could could be called (calling other methods,  e.g. .send , in that mode would result in IllegalStateException being thrown).  If the call is successful (all messages successfully got flushed to all partitions) the transaction is prepared.  If the 2PC is not enabled, we return the INVALID_TXN_STATE error.

...