如何停止接收有关主题的 kafka-metadata 消息答案

【问题标题】：How to stop receiving kafka-metadata messages on a topic如何停止接收有关主题的 kafka-metadata 消息
【发布时间】：2021-08-26 06:22:06
【问题描述】：

出于某种原因，有时，我们的 Confluent Kafka 消费者会收到关于特定主题的奇怪消息：

{
  "magicByte": 2,
  "attributes": 0,
  "timestamp": "1623227829187",
  "offset": "11575814",
  "key": null,
  "value": {
    "type": "Buffer",
    "data": []
  },
  "headers": {},
  "isControlRecord": false,
  "batchContext": {
    "firstOffset": "11575814",
    "firstTimestamp": "1623227829187",
    "partitionLeaderEpoch": 5,
    "inTransaction": false,
    "isControlBatch": false,
    "lastOffsetDelta": 0,
    "producerId": "-1",
    "producerEpoch": -1,
    "firstSequence": -1,
    "maxTimestamp": "1623227829187",
    "timestampType": 0,
    "magicByte": 2
  }
}

即使我吞下了所有错误并在失败后提交了偏移量，我也多次使用了完全相同的消息。

它只发生在一个特定的主题上。我们从 Scarth 删除并重新创建了这个主题，但它没有解决这个问题。

在以下屏幕截图中，我们看到一条消息为空，而其他消息不为空。什么意思？

生产者 - C#：

_producerBuilder = new ProducerBuilder<Null, string>(new ProducerConfig
                {
                    BootstrapServers = _kafka.BootstrapServers,
                    ClientId = Dns.GetHostName(),
                    LingerMs = _kafka.LingerMsKafka,
                    SaslMechanism = SaslMechanism.Plain,
                    SaslUsername = _kafka.SaslUsername,
                    SaslPassword = _kafka.SaslPassword,
                    SecurityProtocol = SecurityProtocol.SaslSsl,
                    EnableSslCertificateVerification = false
                }).Build();

消费者 NodeJ：

this.kafka = new Kafka.Kafka({
  brokers: [`${connectionOptions.host}:${connectionOptions.port}`],
  logLevel: Kafka.logLevel.NOTHING,
  ssl: !!connectionOptions.sasl,
  sasl: connectionOptions.sasl,
})
// ...
this.consumer = this.kafka.consumer({
  groupId: this.groupId,
})
// ...
await this.consumer.connect()
await this.consumer.subscribe({ topic: this.topic, fromBeginning: true })
await this.consumer.run({ batch }: EachBatchPayload) => {/* ...*/})

我们无法使用ccloud 处理有问题的消息：错误 - “恐慌：运行时错误：索引超出范围 [0]，长度为 0”

我们迷路了。任何帮助都会很棒！

NodeJS：v14.15.4 KafkaJS：1.16.0-beta.18

【问题讨论】：

您知道是什么产生了这些消息吗？它是使用事务的生产者吗，因为事务将标记写入没有正文的主题
我不知道是什么产生了这些消息。甚至 confluent 的 kafka-cli 在尝试读取这些消息时都会崩溃。
什么是 kafka 事务？
我相信您能够在其他地方搜索该信息，但它允许幂等生产者和仅一次记录交付
最终，您无法“停止”它，因为写入数据的是生产者。我建议联系 Confluent 支持以了解为什么 UI 能够显示空消息但 CLI 不能。同样，向 KafkaJS 项目提交问题，看看可以做些什么......总的来说，记录有一个唯一的时间戳和偏移量，只是一个空字节数组体，所以没有什么特别之处

标签： node.js apache-kafka confluent-platform confluent-cloud kafkajs

【解决方案1】：

这个问题发生在 kafka confluent (dev) 上。

支持团队没有解决这个问题。

转移到https://aiven.io/，这个问题再也没有出现过。

【讨论】：