从控制台生成具有空值（墓碑）的 Kafka 消息答案

【问题标题】：Producing a Kafka message with a Null Value (Tombstone) from the Console从控制台生成具有空值（墓碑）的 Kafka 消息
【发布时间】：2019-02-03 01:41:47
【问题描述】：

有没有什么方法可以在 kafka-console-producer 中生成一条带有 null 值的消息（即，为压缩器标记它以用墓碑删除它）？

我尝试过制作“mykey”和“mykey|”。前者产生错误，后者使该值成为空字符串。像这样运行生产者：

$KAFKA_HOME/bin/kafka-console-producer --broker-list localhost:9092 --topic mytopic --property "parse.key=true" --property "key.separator=|"

【问题讨论】：

标签： apache-kafka kafka-producer-api

【解决方案1】：

看看kafkacat（kafka 的网猫）。引用文档：

通过提供一个 -Z 解释为 NULL 的空消息值，为键“abc”生成一个墓碑（压缩主题的“删除”）：
echo "abc:" | kafkacat -b mybroker -t mytopic -Z -K:

【讨论】：

仅供参考 kafkacat 目前不支持生成 avro 消息的能力
@MarioP。我每天都这样做：kafkacat -b mybroker -t topic -s avro -r http://schema-registry-url:8080

【解决方案2】：

不幸的是，使用控制台生产者无法做到这一点

这是来自 ConsoleProducer 类的代码 sn-p（它如何读取数据）。 Kafka 0.11.0（不要认为不同版本之间变化很大）。

override def readMessage() = {
  lineNumber += 1
  print(">")
  (reader.readLine(), parseKey) match {
    case (null, _) => null
    case (line, true) =>
      line.indexOf(keySeparator) match {
        case -1 =>
          if (ignoreError) new ProducerRecord(topic, line.getBytes(StandardCharsets.UTF_8))
          else throw new KafkaException(s"No key found on line $lineNumber: $line")
        case n =>
          val value = (if (n + keySeparator.size > line.size) "" else line.substring(n + keySeparator.size)).getBytes(StandardCharsets.UTF_8)
          new ProducerRecord(topic, line.substring(0, n).getBytes(StandardCharsets.UTF_8), value)
      }
    case (line, false) =>
      new ProducerRecord(topic, line.getBytes(StandardCharsets.UTF_8))
  }
}

如您所见，该值始终是不可为空的字节数组

【讨论】：

它在 Kafka 2.0 中已从 Scala 更改为 Java 代码
谢谢娜塔莉亚，看起来你是对的，因为它在最新版本中没有改变。不要认为我可以传递一个 UTF-8 空字符供 getBytes 解析。我有点惊讶他们以这种方式处理“case -1”，因为用户必须明确传递“parseKey”才能进入该块。因此，我认为应该清楚它是传递的键，并且可以将值分配为 null。无论如何，通过改变重建应该很容易。感谢您的帮助！
投票支持增强：issues.apache.org/jira/browse/KAFKA-10238