Kafka Consumer 没有通过 Java 读取主题数据答案

【问题标题】：Kafka Consumer is not reading the topic data through JavaKafka Consumer 没有通过 Java 读取主题数据
【发布时间】：2019-03-08 15:23:38
【问题描述】：

我正在向 Kafka 主题发送输入 JSON 数据。通过使用以下命令，我可以在 kafka 消费者中看到相同的 JSON 数据。

.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic mytopic --from-beginning

案例1： 但是当我试图从 kafka 消费者（java）读取记录时，没有在 java 控制台中获得任何记录。我已经尝试了此链接中给出的示例https://kafka.apache.org/090/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html

案例2： 如果我从生产者（命令窗口）发送任何消息，同样会进入消费者（命令窗口）。并且能够在 java 控制台中看到相同的记录。这种情况正在发挥作用。

如果我通过java程序将数据提交给主题。然后相同的 JSON 数据出现在消费者（命令窗口）中。但是没有进入 java 控制台。Case1 不工作。
Case2 工作。请让我知道需要进行任何配置吗？

【问题讨论】：

所以你开始从 java 消费，然后向主题发送新数据并且没有消息出现（案例 1）？
您是否只使用一个 JVM 运行 case2？
你的制片人是怎么写的？与我们分享您的代码。
同时启动控制台消费者和自定义消费者。然后运行你的 java 生产者。此外，请确保您的生产者向 Kafka 发送消息。你看到两个消费者的消息了吗？

标签： java apache-kafka kafka-consumer-api

【解决方案1】：

你需要确保你没有在最后发送空行并且只使用latest - 像上面一样使用auto.offset.reset: 'earliest'，或者properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

然后， consumer.seekToBeginning(consumer.assignment()); 以确保。

【讨论】：

【解决方案2】：

您需要设置 ConsumerConfig.AUTO_OFFSET_RESET_CONFIG 以便从头开始读取。

kafkaConsumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");

还要确保您没有使用相同的消费者组 ID 运行不同的消费者进程，因为来自一个分区的数据可能会从一个进程读取，而另一个不会看到任何内容。

【讨论】：

【解决方案3】：

假设生产者和消费者代码是正确的。

停止所有消费者。
重置您的消费组

$KAFKA_HOME/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group group_name --topic topic_name --reset-offsets --to-earliest --执行

现在开始您的消费者

这应该可以解决您的问题。

以下是一些 Kafka 消费者属性：

    bootstrap.servers: 'localhost:9092'
    group.id: 'group_id'
    auto.offset.reset: 'earliest'
    key.deserializer: 'org.apache.kafka.common.serialization.*' //Replace * with class
    value.deserializer: 'org.apache.kafka.common.serialization.*'

谢谢

【讨论】：