如何匹配Kafka生产者的记录和RecordMetadata

Question

我想跟踪 kafka 生产者中的记录时间戳：生产者何时发送记录以及代理何时收到确认并存储记录（在确认时间戳上）。 每条记录的标题中都有字符串 id。

对我来说，最好的方法是实现 Kafka ProducerInterceptor：

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

在onAcknowledgement 我有RecordMetadata 并且可以获得acks 时间戳，但我不明白如何匹配我接收元数据的记录？ 正如我从KIP-512 看到的，这是常见问题，正在讨论中。

如何匹配 onAcknowledgement 中的记录？或者可能有更好的方法来跟踪记录的发送/接收时间戳？

不，我不能使用生产者回调，因为我不能更改生产者代码。

Answer 1

这个怎么样？

为生产者发送调用定义一个可以保存所有相关数据的类型：

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

定义一个回调类型，当通过回调方法 onCompletion(...) 可用时将 ProducerSendData 与确认数据耦合起来：

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;
  }
}

然后调用 kafka producer send 方法并使用返回的 future 更新 producerSendData：

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;

有了这个，你应该可以得到所有相关的东西。当然，ProducerSendData.result 未来值可能只在未来可用；）

我希望这是可以理解的。干杯！