KAFKA-9279: Fail producer transactions for asynchronously-reported, synchronously-encountered ApiExceptions

[C0urante]

Jira

Some types of exceptions that the producer reports are encountered synchronously during a call to KafkaProducer::send, but reported asynchronously via the resulting Future instead of being thrown directly from the call to send. One common example of this is when the size of a record exceeds the client-side max.message.bytes configuration property; when that happens, a failed future is returned immediately from KafkaProducer::send.

This is all fine, but the current behavior does not take into account the guarantees surrounding the transactional producer's commitTransaction method, which are that:

...if any of the send(ProducerRecord) calls which were part of the transaction hit irrecoverable errors, this method will throw the last received exception immediately and the transaction will not be committed. So all send(ProducerRecord) calls in a transaction must succeed in order for this method to succeed.

The changes in this PR cause producers to fail on commitTransaction if any prior calls to send resulted in one of these aynchronously-reported, synchronously-encountered exceptions.

An alternative approach could be to throw these exceptions directly to the caller in send, but that would alter producer behavior for all users instead of just ones using transactional producers; that tradeoff seems less desirable just to fix an issue that only affects transactional producers.

Unit tests are added that cover a variety of cases that may lead to previously-missed exceptions and verify that they cause KafkaProducer::commitTransaction to fail as expected.

Committer Checklist (excluded from commit message)

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

[C0urante]

@ableegoldman @guozhangwang @mimaison would any of you be able to take a look?

[mimaison]

Thanks @C0urante for the PR, LGTM!

I left a couple of very minor comments. I leave you the chance to address them if you want, otherwise I'll just merge it in the next couple of days.

[C0urante]

Thanks @mimaison, appreciate it! This should make KIP-618 a little easier to implement (see the note here regarding the workaround I've had to employ right now).