阅读Kafka主题的所有消息
我有以下用例：
我有两个Kafka主题，一个是用来处理传入消息流的，另一个是用来存储记录的，作为应用程序初始状态的引导。
有没有办法做到以下几点：
- 当应用程序启动时，读取Kafka主题中的所有消息，并将该主题中用于将应用程序引导至初始状态的所有
消费者记录存储在内存中 - 只有在读取了所有消息后，才允许处理流主题中的
消费者记录 - 因为在应用程序运行时，状态主题上可能会有其他记录，以便在不必重新启动应用程序的情况下将它们合并到应用程序的状态中
共有1个答案
首先启动引导消费者。
阅读其他主题，直到达到特定的偏移量，或者（如果你想要结束，你可以阅读，只要没有可用的轮询记录[这不是最好的方法！]）。如果你想开始在特定的偏移每次你必须使用一个寻求。也使用唯一的消费者组ID，因为你想要所有的记录。你可能想适当地处理再平衡的情况。
然后关闭该消费者并启动另一个流消费者并处理数据。使用Ktable和Kafka流可能更好，但我不熟悉它。
-
我有一个简单的java制作人，如下所示 我正在尝试读取如下数据 但消费者并没有从Kafka那里读到任何信息。如果我在处添加以下内容 然后消费者开始从题目开始阅读。但是每次消费者重新启动时，它都从我不想要的主题开始读取消息。如果我在启动消费程序时添加了以下配置 然后，它从主题中读取消息，但是如果消费者在处理所有消息之前重新启动，那么它不会读取未处理的消息。 有人可以让我知道出了什么问题，我该如何解决
-
Debezium连接器的Kafka connect事件是Avro编码的。 在传递给Kafka connect standalone服务的connect-standalone.properties中提到了以下内容。 使用这些属性配置Kafka使用者代码: 在消费者实现中，下面是读取键和值组件的代码。我使用REST从模式注册表中获取键和值的模式。 解析密钥工作正常。在解析消息的值部分时，我得到了Arr
-
我正在使用kafka:Kafka2.12-2.1.0，在客户端使用spring kafka，但遇到了一个问题。 我需要通过阅读Kafka主题中的所有现有消息来加载内存中的映射。为此，我启动了一个新的使用者（具有唯一的使用者组id，并将偏移量设置为)。然后我迭代使用者（poll方法）以获取所有消息，并在使用者记录变为空时停止。 我尝试了很少的其他方法（比如使用偏移量数），但还没有找到任何解决方法，除
-
我使用confluent .net客户端。订阅者在重启（订阅者服务重启）后始终读取 Kafka 主题的所有消息。如何提交消费者已经实现的偏移并从中读取？也许一些消费者配置可以提供帮助...
-
因为我是新的Kafka，所以我能够从文件中读取记录，并通过生产者将消息发送到Kafka主题，但不能通过消费者消费相同的主题。 注意：您可以从任何文本文件中读取数据，我使用的是Kafka2.11-0.9。0.0版本 这是我的密码： 下面是输出：
-
我试图消费一个Kafka主题从Spring启动应用程序。我使用的是下面提到的版本的Spring云流 Spring boot starter父级：2.5.7 Spring云版本：2020.0.4 下面是代码和配置 application.yml 消息消费者类 下面的消息发布者正在正确地发布消息。发布者是在不同的微服务中编写的。 pom.xml