KAFKA-8351; Cleaner should handle transactions spanning multiple segments

[hachikuji]

When cleaning transactional data, we need to keep track of which transactions still have data associated with them so that we do not remove the markers. We had logic to do this, but it was not being carried over when beginning cleaning for a new set of segments. This could cause the cleaner to incorrectly believe a transaction marker was no longer needed. The fix here carries the transactional state between groups of segments to be cleaned.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[dhruvilshah3]

Thanks for the patch. LGTM, left couple of minor comments.

[viktorsomogyi]

I don't see any problem with this fix but have some further thoughts:

• What if the compaction policy is "delete"?
• Are transaction markers kept in sync between user topics and __consumer_offsets and __transaction_state with regard to cleaning?

[viktorsomogyi]

nit: might be better to use addAbortedTransactions.

[junrao]

@hachikuji : Thanks for the PR. LGTM. Just a minor comment below.

[junrao]

Could we add transactionMetadata to the javadoc?

[hachikuji]

@viktorsomogyi Thanks for the questions.

What if the compaction policy is "delete"?

Just like a normal topic, producer state can be lost due to retention. Note that I have proposed in KIP-360 to retain producer state beyond the retention in the log. That may allow us simplify the cleaner's bookkeeping here.

Are transaction markers kept in sync between user topics and __consumer_offsets and __transaction_state with regard to cleaning?

I'm not 100% sure I understand the question, but transaction state is local to the partition after the transaction completes. Once markers are written to the log, the transaction is considered complete locally and the data is exposed.