我的代码看起来像这样。message_reader(process_consumer) process_consumer.close()
所以，我的问题是，假设主题中没有消息，也没有消息被消费- commit()会增加偏移量吗？另外，生产者是否在生成消息之前检查最新的偏移量？
浏览 15提问于2020-03-05得票数 0
1回答
在异常的应用程序终止并重新启动之后，我的应用程序开始接收来自Kafka队列的旧的、已经处理过的消息。spring.kafka.bootstrap-servers=${kafka.host}:${kafka.port}
spring.kafka.consumer.auto-offset-reset=ea
浏览 0提问于2018-07-08得票数 4
回答已采纳
1回答
但是，我遇到了一个我不知道如何处理的用例。这个场景是一个流，它的源是ActorPublisher，它消耗的是Kafka的消息，而订阅者是更新Cassandra表的接收器。Kafka ~>一些映射操作~> Cassandra
关键是，每当一条消息成功地被处理并插入卡桑德拉时，我都想向卡夫卡明确确认，这样我就可以在灾难发生和服务失败的情况下重新读取这条<e
浏览 2提问于2015-05-27得票数 4
1回答
速率限制:由于Kafka能够以远高于MQ消耗的速率生成消息，我们是否可以使用一些配置设置@ kafka consumer来启用速率限制传输，以保护MQ的稳定性？同样-只需一次语义-理解kafka只支持一次语义，这将阻止已经被消费者消费的消息的重新传输。有人可以指导我如何设置此配置吗？
我们在组织中使用的是confluent kafka企业版。
浏览 1提问于2020-10-07得票数 0
我试图使用gpss (格林梅利流服务器)将数据从Kafka加载到GreenplumDB。使用给定的config)Write (在kafka
浏览 5提问于2021-01-13得票数 0
回答已采纳
现在只是为了测试，只有一个主题，对于这个主题，1个使用者正在从主题和处理中获取相同的消息，并将它们存储在数据库中。但是，如果我在数据库中存储和抛出异常(例如PersistenceException )有任何问题，那么消息流就会被中断。我该怎么处理呢？有人面对这种情况吗？
浏览 1提问于2021-09-13得票数 1
回答已采纳
如果我有一个Kafka流应用程序无法发布到主题(因为主题不存在)，它是提交消费者偏移量并继续，还是在相同的消息上循环，直到它可以解析输出主题？尝试发布到主题时的错误示例： Error while fetching metadata with correlation id 80 : {super.cool.test.topic=UNKNOWN_TOPIC_OR_PARTITION} 在我看来，为了不丢失数据，它只会在相同的消息上旋转，直
浏览 13提问于2019-05-31得票数 1
回答已采纳
我正在使用Spark2.2.0和Kafka0.10火花流库从满是kafka应用程序的主题中读取。Kafka Broker版本为0.11，Kafka-streams版本为0.11.0.2。当我在Kafka应用程序中设置EXACTLY_ONCE保证时：
p.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE编辑1:主题填充卡夫卡-流应用程序(完全一次启用)有错
浏览 0提问于2018-01-19得票数 2
回答已采纳
3回答
我有一个弹簧启动应用程序与单一的kafka消费者，以获取一些主题的消息。但有时在处理消息时会发生错误。我希望像往常一样继续接收以下消息，同时能够不丢失该消息并接收它，例如，下次服务在修复后与使用者重启时。
有可能做到这一点吗？我知道我需要禁用自动提交并手动提交成功的消息，但是，在这种情况下，如果我没有对这个异常
浏览 0提问于2020-08-30得票数 1
有一个消息到达的Kafka主题。我需要阅读一条消息，对其进行处理，然后继续下一条消息。消息处理可能会失败，如果发生这种情况，处理必须重试几次(比方说10次)，然后才能继续处理下一条消息。如果处理失败10次，则需要删除该消息，我们应该继续处理下一条消息。 我们使用reactor-kafka，所有的处理都需要反应式的。要提交偏移量，我们需要一个ReceiverReco
浏览 23提问于2020-04-09得票数 2
我正在尝试使用kafka-node来读取来自kafka主题的压缩消息。kafka-topics.sh --zookeeper ${K
浏览 2提问于2018-08-25得票数 0
它还必须以接收消息的相同顺序发出它们。这意味着flow不能过滤或收集消息的子集，相反，它必须将消息拆分成单独的流，并将那些本应被丢弃的消息映射到Done。此外，在Lagom的KafkaSubscriberActor的实施中，我们看到私有atLeastOnce的实施本质上是解压缩消息有效负载和偏移量，然后在我们的用户流将消
浏览 25提问于2019-03-12得票数 3
2回答
我是Apache Nifi的新手，但我在尝试连接kafka微服务(与生产者)和Apache nifi消费者时遇到了一些问题。/test depends_on: command: python test.py 测试是我的制作人： from在Apache nifi仪表板中，我使用以下参数设置了一个输入: Kafka broker: localhost:9
浏览 37提问于2021-07-14得票数 0
2回答
我是卡夫卡的新手。我的问题是：，我们想知道如何重新使用未提交的消息？捕获应用程序异常并跳过kafka提交。将消息存储到RAM中，然后重试。
在这种情况下的最佳实践方法是什么</
浏览 2提问于2018-03-21得票数 2
我在问自己，spring-kafka项目的ReactiveKafkaConsumerTemplate是否保证了消息的正确排序。我阅读了reactor-kafka项目的文档，它指出消息应该使用concatMap操作符，但ReactiveKafkaConsumerTemplate至少在这里使用receiveAutoAck方法的情况下使用/docs/kafka/release/ref
浏览 37提问于2021-01-26得票数 0
回答已采纳
我正在设计一个apache storm拓扑(使用streamparse)，它使用一个spout (apache kafka spout)和一个并行大于1的螺栓构建，可以批量读取kafka spout中的消息并将消息持久化到mysql表中当螺栓在mysql上插入失败时，我不会在kafka中提
浏览 2提问于2019-01-31得票数 0
1回答
我正在开发一个使用Kafka流的spring boot应用程序，在我的应用程序中，我希望管理Kafka偏移量，并在仅在成功处理消息的情况下提交偏移量。这一点很重要，确保即使Kafka重启或zookeeper关闭，我也不会丢失消息。我目前的情况是，当我的Kafka断断续续时，我的消费者从头开始并使用所有之前<em
浏览 3提问于2020-01-16得票数 0
我写了一个node-kafka consumer。在极少数情况下，我使用一个组id启动kafka客户端，当某些偏移量可用时使用，但不再可用-导致"offsetOutOfRange“事件被调用。在这种情况下，推荐的行为是什么？记录错误并退出？有没有办法恢复？我总是想从上次提交的偏移量开始运行zookeeper (如果存在且可用)。client = new kafka.Cli
浏览 10提问于2016-08-01得票数 1
回答已采纳
我刚接触Kafka，我并不真正理解Kafka配置的含义，有没有人能给我解释一下更容易理解！下面是我的代码： "bootstrap.servers" -> "master:9092,slave1:9092",auto.offset.reset" -> "latest", //earliest or latest
&quo
浏览 0提问于2017-07-05得票数 21
1回答
将此操作并行化的最佳方法是什么？例如，假设我创建了1000个分区。我考虑过让http请求是异步的，但在确定要提交的偏移量时，我似乎遇到了问题。
例如，在
浏览 3提问于2015-05-08得票数 1
回答已采纳
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
