我有一个spring引导服务，它使用kafka主题。当我消费时，我会对kafka消息执行某些任务。在执行这些操作之前，我需要等待服务将一些数据加载到我设置的缓存中。我的问题是，如果我将kafka consumer设置为自动启动，它会在缓存加载之前开始消耗，然后它就会出错。
我试图在加载缓存后显式地启动使用者，但是我得到了空指针异常。("${kafka.p
浏览 45提问于2019-11-07得票数 0
我正在按照快速入门指南上的说明在本地运行kafka。然后，我在config/consumer.properties中定义了我的使用者组配置，以便我的使用者可以从定义的group.id中挑选消息bin/kafka-consumer-groups.shconsole-consumer-67807 <-- when connecting to kafka
浏览 1提问于2020-05-13得票数 5
回答已采纳
在我重新启动Linux之前，一切都很好。重新启动后，当我尝试使用kafka-console-consumer.sh使用队列时，会出现以下错误。$ ~/kafka/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic TutorialTopic --from-beginning2016-02-04 :16:54,944警告conso
浏览 9提问于2016-02-04得票数 5
spring.cloud.stream.bindings.input.group=sampleconsumergroup消费者服务听取主题模式。只有返回联机后发送的消息才会被消耗。
这是否如预期的那样？我正处于
浏览 2提问于2019-03-06得票数 1
回答已采纳
我正在尝试设置一个Kafka集群(实际上是集群中的第一个节点)。当我启动kafka时，我得到了以下异常(导致kafka关闭)。编辑:我启动了kafka，我必须在server.config文件中</em
浏览 1提问于2014-08-26得票数 18
回答已采纳
在下面的Linux事务片段中，我用myserverA、myserverB和myserverC替换了我的服务器名称。我还模糊了密码。这是我第一次在留言板上发帖。我为这篇文章中任何格式不佳的部分道歉。我可以通过SSL使用kafka-console-producer生成数据，但不能使用kafka-console-consumer读取数据。myserverA:9093 --zookeeper myserverA:2181/kafka --topic ssl-test
浏览 1提问于2017-06-28得票数 3
我一直在尝试与kafka-avro-console-consumer从Confluent连接到我们的遗留Kafka集群，该集群是在没有Confluent Schema Registry的情况下部署的。我使用如下属性显式地提供了模式： --topic test \org.apach
浏览 2提问于2018-09-19得票数 6
回答已采纳
我使用的是Kafka Stream API。当我启动我的应用程序时，有时会有一个间隙，我想从一个特定的偏移量开始消费。最早或最晚的不是我想要的。streamProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); 我正在寻找的是一个场景，比如我在配置文件中设置了偏移量或以毫秒为单位的日期
浏览 25提问于2020-05-02得票数 2
回答已采纳
当消费者实例group.id在重启时更改时，是否会对kafka性能产生任何影响。较旧的group.id将会发生什么情况?它是否仍然存在于代理内存中，或者何时将被删除？
浏览 8提问于2018-08-28得票数 0
回答已采纳
kafka-consumer-groups.bat --bootstrap-server kafka-host:9092 --group my-group --reset-offsets --to-earliestNEW-OFFSETSecondTopic 0 0kafka-consumer-groups.bat
浏览 2提问于2017-08-14得票数 47
回答已采纳
在这样做的过程中，我们的测试现在都无法启动，原因是：
org.apache.kafka.clients.NetworkClient : [Consumer clientId=consumer-1,扩展，该扩展在测试之间停止/启动容器侦听器(同时将所有分区和主题查找到最新，以防止测试意外泄漏消息，从而在以后的测试中打破预期)。在上一个版本中，这不是一个问题；我们会看到一些日志消息，说明在扩展配置和
浏览 0提问于2019-05-03得票数 4
回答已采纳
2回答
@SpringBootTest注释的所有可能的测试中都是活动的，而不仅仅是在卡夫卡特有的测试中。是否有可能停止每个测试的所有@KafkaListener，并为特定的@SpringBootTest显式启用它们，@KafkaListener附带了一个卡夫卡模块。这一款采用了融合卡夫卡码头形象，在配置上非常顽固。例如，您不能设置一些代理属性，也不能告诉容器应该在启动后创建哪个主题。后一种方法的问题是，当我使用命令行maven运行</
浏览 7提问于2020-06-29得票数 1
我使用kafka节点(kafka的节点客户端)，使用使用者检索有关主题的消息。不幸的是，我收到了一个" offsetOutOfRange“条件(调用offsetOutOfRange回调)。我的应用程序运行良好，直到消费者明显落后于生产者，在最早的和最新的抵消之间留下了很大的差距。此时，我(可能是错误的)假定使用者将能够继续接收消息(并希望赶上生产者)。我的kafka客户端代码如下：:
var kafka = require('kafka-nod
浏览 4提问于2015-08-24得票数 4
回答已采纳
我是kafka的新手，正在尝试理解是否有一种方法可以从上次消费的偏移量读取消息，而不是从开始读取。1) I produced 5 messages at 7:00 PM and console consumer consumed those.4) Now, i have started console consumer<
浏览 0提问于2015-11-13得票数 15
回答已采纳
3回答
我有一个用例，其中有一个消费群体正在使用消息。我想构建一个API来修改它的偏移量。因此，当使用偏移量调用端点时，我必须更改使用者组的偏移量。我使用的是SpringBoot，消费者是用SpringBoot构建的。提前谢谢。
浏览 4提问于2021-04-16得票数 2
回答已采纳
我使用pykafka从kafka主题中获取消息，然后进行一些处理并更新到mongodb。因为pymongodb每次只能更新一项，所以我启动了100个进程。但在启动时，一些进程出现错误"PartitionOwnedError和ConsumerStoppedException“。我也不知道原因。谢谢。topic = kafka_client.topics[topic_name]
balanced_consumer = topic.get_balanced_
浏览 5提问于2016-09-21得票数 3
6回答
我在使用KafaConsumer时遇到了麻烦，无法让它从头开始读取，或者从任何其他显式偏移量读取。针对同一主题为使用者运行命令行工具时，我确实看到带有--from-beginning选项的消息，否则它将挂起如果我通过python运行它，它会挂起，我怀疑这是由不正确的使
浏览 1提问于2016-02-05得票数 12
="#{T(org.apache.kafka.clients.consumer.ConsumerConfig).KEY_DESERIALIZER_CLASS_CONFIG}"value="org.apache.kafka.common.serialization.StringDeserializer"/>
<entry key="#{T(org.apache.kafka.clients.consumer.Consume
浏览 19提问于2019-03-27得票数 0
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
