从控制台生成具有空值(墓碑)的 Kafka 消息

Dav*_*apy 4 apache-kafka kafka-producer-api

有什么方法可以在 kafka-console-producer 中生成一个带有空值的消息(即标记它以便压缩器用墓碑删除它)?

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

$KAFKA_HOME/bin/kafka-console-producer --broker-list localhost:9092 --topic mytopic --property "parse.key=true" --property "key.separator=|"
Run Code Online (Sandbox Code Playgroud)

mat*_*ieu 5

看看kafkacat(kafka的netcat)。引用文档:

通过提供 -Z 解释为 NULL 的空消息值,为键“abc”生成墓碑(对于压缩主题的“删除”):

echo "abc:" | kafkacat -b mybroker -t mytopic -Z -K:
Run Code Online (Sandbox Code Playgroud)

  • @马里奥P。我每天都这样做:`kafkacat -b mybroker -t topic -s avro -r http://schema-registry-url:8080` (3认同)

Nat*_*lia 2

不幸的是,没有办法使用控制台生产者来做到这一点

这是 ConsoleProducer 类的代码片段(它如何读取数据)。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))
  }
}
Run Code Online (Sandbox Code Playgroud)

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