由于 Confluent Interceptor，Kafka Connect 跳过消息答案

【问题标题】：Kafka Connect Skipping Messages due to Confluent Interceptor由于 Confluent Interceptor，Kafka Connect 跳过消息
【发布时间】：2022-12-03 10:31:36
【问题描述】：

我在连接日志中看到以下消息

WARN Monitoring Interceptor skipped 2294 messages with missing or invalid timestamps for topic TEST_TOPIC_1. The messages were either corrupted or using an older message format. Please verify that all your producers support timestamped messages and that your brokers and topics are all configured with log.message.format.version, and message.format.version >= 0.10.0 respectively. You may also experience this if you are consuming older messages produced to Kafka prior to any of those changes taking place. (io.confluent.monitoring.clients.interceptor.MonitoringInterceptor)

我用这个改变了我的卡夫卡经纪人

KAFKA_INTER_BROKER_PROTOCOL_VERSION: 0.11.0
KAFKA_LOG_MESSAGE_FORMAT_VERSION: 0.11.0

我猜这会降低我的整体生产者吞吐量，我正在尝试进行负载测试。

附言：我不想删除 confluent 拦截器，因为它可以帮助我解决吞吐量和消费者滞后问题。

CONNECT_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor"
CONNECT_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor"

任何不跳过这些消息的方法，我正在使用pepperbox来生成消息并且它没有时间戳

{
        "messageId":{{SEQUENCE("messageId", 1, 1)}},
        "messageBody":"{{RANDOM_ALPHA_NUMERIC("abcedefghijklmnopqrwxyzABCDEFGHIJKLMNOPQRWXYZ", 2700)}}",
        "messageCategory":"{{RANDOM_STRING("Finance", "Insurance", "Healthcare", "Shares")}}",
        "messageStatus":"{{RANDOM_STRING("Accepted","Pending","Processing","Rejected")}}"
}

提前致谢！

【问题讨论】：

标签： apache-kafka apache-kafka-connect confluent-platform

【解决方案1】：

查看 pom 中的 Kafka 版本，您会看到它使用的是 Kafka 0.9

从 0.10.2 开始，时间戳被添加到 Kafka 中。

正如错误所述Please verify that all your producers support timestamped messages。

使用新版本重新编译项目，所有生成的记录都会自动带有时间戳，因此不会被跳过。

或者，使用不同的工具，如 JMeter 或 Kafka Connect Datagen。

【讨论】：