...

This can happen when a message gets stuck or delayed due to networking issues or a network partition, the transaction aborts, and then the delayed message finally comes in. The delayed message case can also violate EOS if the delayed message comes in after the next addPartitionsToTxn request comes in. Effectively we may see a message from a previous (aborted) transaction become part of the next transaction. A similar event can occur if a delayed EndTxn marker comes through an a new transaction with the same epoch has started when the request is received. This can result in incorrectly completing a transaction.

Another way hanging transactions can occur is that a client is buggy and may somehow try to write to a partition before it adds the partition to the transaction. In both of these cases, we want the server to have some control to prevent these incorrect records from being written and either causing hanging transactions or violating Exactly once semantics (EOS) by including records in the wrong transaction.

...

For more detail – consider the epoch bump cases for the transactional state records and EndTxn responses where we do and do not see epoch overflow:

Normal EndTxn (not epoch overflow)

Prepare

• Keep producer ID but increment epoch

• Return new epoch and producer ID in EndTxnResponse

• If we retry and see epoch - 1 + ID in last seen fields and are issuing the same command (ie commit not abort) we can return (with the new epoch)

Complete

• Write the previous ID in the tagged field. (Epoch - 1 is assumed)

• If we retry and see epoch - 1 + ID in last seen field and are issuing the same command we can return (with new epoch)

Epoch Overflow EndTxn

Prepare

Say we have producer ID x and epoch y. When we overflow epoch y we get producer ID z.

PREPARE
producerId: x
*previous/lastProducerId (tagged field): x
nextProducerId (tagged field): empty or z if y will overflow
producerEpoch: y + 1

• Non-overflow: Return epoch y + 1 and producer ID x

• Overflow: Return epoch 0 and producer ID z for overflow in EndTxnResponse

• Keep the previous producer ID field as x. We use this field to signify that a new client + new server (with epoch bump) set the field.

• Write the producer ID and max epoch in the producer ID and epoch fields. Get a new producer ID and write it in the nextProducerId field. The previous producer ID tagged field will remain empty.

• Return epoch 0 and new producer ID in EndTxnResponse

• If we retry and see epoch - 1 + and producer ID x in last seen fields and are issuing the same command we can return ((ie commit not abort) we can return (with the new producer id and epoch)

Complete

COMPLETE
producerId: x or z if y overflowed
*previous/lastProducerId (tagged field): x
nextProducerId (tagged field): empty
producerEpoch: y + 1 or 0 if we overflowed

• Non-overflow: Set the producer ID to x

• Overflow: Set producer ID to z (from nextProducerId field), set nextProducerID and epoch to 0, and nextProduceId can go back to empty. Set

• Keep the previous producer ID field to x.

• If we retry and see epoch max - 1 + ID in last seen fields and are issuing the same command we can return (with new current producer id ID and epoch)

Once we move past the Prepare and Complete states, we don’t need to worry about lastSeen fields tagged fields and clear them, just handle state transitions as normal.

In KIP-890 part 2, when writing to a partition we will check the epoch when we add the partition, and use the existing fencing logic at the log layer (and we do not worry about the verification mechanism described in part 3).

*NOTE: When loading from the log and previous/lastProducerId field is present we can populate the lastProducerId and lastProducerEpoch fields.

Return Return Error for Non-Zero Sequence on New Producers

...

Unit/Integration testing will be done to test the various hanging transaction scenarios described above. Tests will also be done to ensure client compatibility between various versions.

Rejected Alternatives

In the overflow case, have a CompleteCommit record with the old producer ID + the new producer ID written in a pseudo InitProducerIdRecord

This is closer to how the old client works in the epoch overflow case. We could also atomically commit both records. However, since the records are cleaned up via compaction with the transactional ID as the key, we can lose information about relationships between producer IDs. 

Although the complete record with new producer ID is a bit strange seeming, it is best for when the server downgrades and the client has the new producer ID. It also doesn't have any problems with correctness.

Bump the epoch on EndTxn but write the marker with the old epoch

...