生产记录到 Kafka答案

【问题标题】：Produce Record to Kafka生产记录到 Kafka
【发布时间】：2020-05-24 06:39:22
【问题描述】：

我正在尝试在不传递任何分区值但想发送标头的情况下向 Kafka 生成记录，并且我有以下构造方法来向 Kafka 生成记录：

ProducerRecord(java.lang.String topic, java.lang.Integer partition, java.lang.Long timestamp, K key, V value)

创建具有指定时间戳的记录，发送到指定主题和分区

ProducerRecord(java.lang.String topic, java.lang.Integer partition, java.lang.Long timestamp, K key, V value, java.lang.Iterable<Header> headers)

创建具有指定时间戳的记录，发送到指定主题和分区

ProducerRecord(java.lang.String topic, java.lang.Integer partition, K key, V value)

创建要发送到指定主题和分区的记录

ProducerRecord(java.lang.String topic, java.lang.Integer partition, K key, V value, java.lang.Iterable<Header> headers)

创建要发送到指定主题和分区的记录

ProducerRecord(java.lang.String topic, K key, V value)

创建要发送到 Kafka 的记录

ProducerRecord(java.lang.String topic, V value)

创建没有键的记录

在上述所有方法中，我无法在不发送分区值的情况下发送标头，如果我将分区设置为 null，我将得到 NullPointerException。

您能否建议我如何通过发送标头而不是分区值在 Kafka 中生成记录。

【问题讨论】：

你能分享一下 NullPointerException 堆栈跟踪吗？应该可以将partiton作为null传递。

标签： apache-kafka kafka-producer-api

【解决方案1】：

ProducerRecord 对象有方法headers()，它返回Headers 接口，这反过来又允许你add() 任何你喜欢的标题。

【讨论】：

【解决方案2】：

Kafka ProducerRecord Instance 提供了 Headers Instance，我们可以在其中添加键值对。

您可以浏览 Kafka 源代码以获取完整信息。

https://github.com/a0x8o/kafka/blob/master/clients/src/main/java/org/apache/kafka/clients/producer/ProducerRecord.java

示例

我尝试使用kafka-clients-0.8.2.0，但该功能在该版本中不可用。所以你还需要检查 Kafka 的客户端版本。

下面给出的例子是使用Kafka Client - 2.1.0。

        ProducerRecord<String, byte[]> record = new ProducerRecord<>(topic, key, message);
        Headers headers = record.headers();
        headers.add(KafkaHeaders.HEADER_CLIENT_IP, Strings.bytes(Network.localHostAddress()));
        headers.add(KafkaHeaders.HEADER_CLIENT, Strings.bytes(logManager.appName));
        producer.send(record);

【讨论】：