How to match Kafka producer's record and RecordMetadata

user_16788454 0
Viewed 324

I want to track records timestamps in kafka producer: when record was sent by producer and when broker receive acks and store record (on ack timestamp). Each record has string id in headers.

For me the best way for it is implement Kafka ProducerInterceptor:

    @Override
    public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
    }
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
    }

In onAcknowledgement i have RecordMetadata and can get acks timestamp, but i can't understand how i can match for which record i receive metadata? As I can see from KIP-512 it's common problem, and under discussion.

How I can match record in onAcknowledgement? Or may be there is a better way to track record's send/receive timestamps?

And no, I can't use Producer callback, cause i can't change producer code.

kafka-producer-api apache-kafka
1 Answers

How about this?

Define a type that can hold all relevant data for a producer send call:

class ProducerSendData<K, V>
{
  ProducerRecord<K, V>   parameter;
  Future<RecordMetadata> result;
  RecordMetadata         acknowledgeMetaData;
  Exception              acknowledgeException;
}

Define a Callback type that couples ProducerSendData with acknowledgement data when this is available via callback method onCompletion(...):

class ProducerSendDataCallback<K, V> implements Callback
{
  // constructor injection ...
  ProducerSendData<K, V> producerSendData;

  /** updates {@link #producerSendData} with method acknowledgement metadata and exception */
  @Override public void onCompletion(RecordMetadata metadata, Exception exception)
  {
    producerSendData.acknowledgeMetaData  = metadata;
    producerSendData.acknowledgeException = exception;
  }
}

Then call kafka producer send method and update producerSendData with returned future:

ProducerSendData               producerSendData = new ProducerSendData(new ProducerRecord(whatever);

ProducerRecord                 methodParameter  = producerSendData.parameter;
ProducerSendDataCallback<K, V> methodCallback   = new ProducerSendDataCallback<>(producerSendDataItem);
Future<RecordMetadata>         methodResult     = producer.send(methodParameter, methodCallback);

producerSendData.methodResult = methodResult;

With this you should have available everything that is relevant. ProducerSendData.result future value may only be available in the future, of course ;)

I hope this is comprehensible. Cheers!

user_65353840
Related