KAFKA-13448: Align behaviour for producer callbacks with documented behaviour

[soceanainn]

DO NOT MERGE Awaiting discussion and acceptance of KIP-799 first

Originally, Callback would return a null metadata value when an error occurred.

This was partially changed by KAFKA-3303, where in some cases Callback would return an 'empty' metadata. In this empty metadata TopicPartition is set correctly but all other fields are set as -1.

The docs were later updated by KAFKA-7412, but it incorrectly states that Callback will always return this 'empty' metadata when an error occurs. However in the case of any exceptions that are a subclass of ApiException, Callback will still return a null value (see here).

This change fixes the behaviour when an ApiException is thrown, to align that behaviour with other exceptions and with the currently documented behaviour.

Changes

• In the one case where KafkaProducer was calling Callback.onCompletion directly, update the call to pass a placeholder value for metadata (in a manner consistent with the Javadoc) instead of a null value. Add unit tests around this change.
• Move the private nested InterceptorCallback class out of KafkaProducer, and create a dedicated public class for InterceptorCallback within 'clients.producer.internal' instead.
• Where KafkaProducer was passing InterceptorCallback to methods within 'clients.producer.internal' classes, update those methods to expect InterceptorCallback as input instead of Callback. Update unit tests as appropriate.

Committer Checklist (excluded from commit message)

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

[soceanainn]

@junrao can you grant permission for me to create a KIP for this change?

[junrao]

@soceanainn : What's your wiki ID? Thanks.

[soceanainn]

@junrao it is seamus.oceanainn

[junrao]

@soceanainn : Just gave you the wiki permission.

[soceanainn]

@junrao thanks, I'll try to create a KIP tomorrow.

I've added a test for the ApiException case, however for the other case (using InterceptorCallback), I can't find a way to test it from the KafkaProducer class.

It's easier to test from the 'ProducerBatch' side of things, but this leads to complications since it is testing that metadata is correctly set to null (e.g. here).

The easiest fix is to make InterceptorCallback accessible at the package level, and update the ProducerBatch method to accept InterceptorCallback as input instead of Callback. That way it would make this implicit conversion from null metadata to an 'empty' record more obvious, and allow us to write proper unit tests around this. What do you think?

[junrao]

@soceanainn : Thanks for the KIP. Making ProducerBatch accept InterceptorCallback seems fine since it's an internal class.

[soceanainn]

An equivalent change was implemented and merged in #11689