KAFKA-5913: Add hasOffset() and hasTimestamp() methods to RecordMetadata

[apurvam]

We return this exception from RecordMetadata.offset() or RecordMetadata.timestamp() if these pieces of metadata were not returned by the broker.

This will happen, for instance, when the broker returns a DuplicateSequenceException.

[apurvam]

cc @hachikuji @ijuma

I'm open to suggestions for more places where good test cases can be added, if any.

[tedyu]

lgtm

[ijuma]

One thing to think about is that the ProducerInterceptor currently has:

* @param metadata The metadata for the record that was sent (i.e. the partition and offset).
     *                 If an error occurred, metadata will contain only valid topic and maybe
     *                 partition. If partition is not given in ProducerRecord and an error occurs
     *                 before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
     *                 The metadata may be null if the client passed null record to
     *                 {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
     * @param exception The exception thrown during processing of this record. Null if no error occurred.

So, a RecordMetadata with missing data relies on sentinels for some of the fields. I'm not a fan of that approach, but it seems a bit weird for some fields to throw an exception while others throw a sentinel. I haven't thought it in depth though, so maybe there's a good reason to do it that way.

[apurvam]

retest this please

[apurvam]

@ijuma that documentation only applies to cases where an error occurred. In the duplicate sequence case, we don't return an error, but don't return the record metadata either. Perhaps a compromise could be that we only raise the RecordMetadataNotAvailableException in cases where there is no error set, but where we don't have a valid offset and timestamp?

This would be complementary to the expectations of the producer interceptor.

[ijuma]

Seems like this needs to be updated.

[hachikuji]

LGTM. I will merge with a couple trivial cleanups (including the one mentioned by @ijuma).