我使用@KafkaListener来消费来自kafka主题，我有一个应用程序逻辑来处理不同消费者组中的多个消费者的每个记录。 现在我的问题陈述是，-I必须将使用的消息发送到第三方rest端点。如果消息发送到rest-end点失败，则不应提交偏移量，需要根据可配置的次数重试发送消息。在可配置的重试</em
浏览 83提问于2019-02-12得票数 2
回答已采纳
我正在从基于JMS的系统转移到Kafka，Kafka负责在新系统和旧系统之间同步事务。新系统发布的每条消息都必须由订阅者/消费者成功处理。我不必担心消息的顺序。由于遗留系统中的一些设计问题(悲观锁定)，偶尔会有一些事务在消息到达时失败，在这种情况下，我希望消息在一段时间后返回。我正在试图弄清楚如何处理与Kafka的这种情况。我的源应用程
浏览 2提问于2017-05-06得票数 0
1回答
我正在开发一个模块，在这个模块中，它使用来自Kafka主题的消息并发布到下游系统。在下游系统不可用的情况下，消费者不要确认Kakfa消息。因此，当我的消费者在下游系统不可用时收到消息时，kakfa的偏移量将不会被提交。但如果我在下游系统启动后收到新消息，并且当我确认该消息时，最新的偏移量将被提交，消费者将永远不
浏览 17提问于2019-08-01得票数 0
我有一个要求消费从kafka主题，做一些工作的记录，并产生到另一个主题与spring-kafka 2.1.7。其他要求是事务性的只有一次语义，重试和错误handling.On提交记录失败我应该做3次重试，记录每个重试消息重试主题和失败的所有重试发送记录到一个死信主题。我看过，它有很好的细节来解
浏览 5提问于2018-08-02得票数 0
回答已采纳
我正在为基于Kafka的数据处理管道中的消费者重试提出一个架构。我们正在使用Kafka生产者和消费者，并正在考虑重试主题，如果他们在消费时出错，将发送消息。将会有消费者以一定的节奏运行在这些重试主题上。我们的</
浏览 1提问于2018-07-24得票数 2
3回答
我正在使用开发，并观察到一些让我怀疑自己是否做错了什么或者它是否按预期工作的行为--我对此毫不怀疑：我的设置尽可能简单。一个单独的Kafka代理和一个只有一个分区的主题。具有默认设置的代理、主题、生产者和使用者(auto为true)。测试案例1
浏览 5提问于2017-10-12得票数 3
有没有可能在SpringBoot上为每个卡夫卡主题的消费者设置不同的投票率？我希望有一个主题是轮询与较长的间隔(如5分钟)。我希望使用它来实现对来自kafka的消息的失败处理的重试。
浏览 106提问于2020-08-20得票数 0
回答已采纳
3回答
我使用的是Confluent.Kafka .NET客户端版本1.3.0。// log }问题是，即使我注释掉行consumer.StoreOffset(consumerResult);，我也会在下一次使用时得到下一条未使用的消息，即偏移量不断增加--这似乎不是文档声称的那样，至少有一次传递。，即即使我注释掉了Commit，我仍然会得到下一个未使用的消息。我觉得我在这里错过了一些基本的东西，但我想不出
浏览 12提问于2020-03-08得票数 17
1回答
我正在处理kafka消息，并使用spark streaming和手动偏移量提交插入到kudu表中。这里是我的代码。commit}
让我们在向kudu中插入数据时列出此示例，我得到了再次处理这些消息所需的错误，如果我停止作业并再次启动它，我能够获得未提交的消息我们不能在流中获得所有未提交的消息吗？
浏览 0提问于2019-03-28得票数 0
我使用和Spring函数创建了一个Kafka消费者，用于批量消费来自Kafka主题的消息。现在，我希望将错误批次发送到死信队列中，以便进一步调试错误。我正在用Spring重试处理我的使用者方法中的重试。但是对于不可重试的异常，我希望将整个批处理发送到DLQ。这就是我的消费者的样子：
@Bean
浏览 9提问于2022-07-11得票数 0
1回答
需要指导如何为Kafka消费者类编写集成JUnit测试，以断言消息消费被重试了多少次。我在Kafka配置中添加了一个重试模板。retryPolicy.setMaxAttempts(3); return retryTemplate;消费级@KafkaListener(topics = "${kafka.topic}&qu
浏览 2提问于2021-07-15得票数 0
1回答
为了消费来自Kafka的数据，我们可以在一个主题上有多个消费者，完全解耦。那么，在卡夫卡和脉冲星之间存在差异的页面上的no shared consumption (https://streaml.io/blog/pulsar-streaming-queuing)是什么意思？
浏览 19提问于2019-04-26得票数 2
回答已采纳
1回答
我使用的是Spring Kafka consumer，它从主题中获取消息并将其持久化到数据库中。如果满足故障条件，例如数据库不可用，kafka消费者库是否提供重试机制？如果是这样，有没有办法设置不同的重试间隔，比如第一次重试应该在5分钟后完成，第二次重试应该在30分钟后完成，第三次重试应该在1小时后完成，等等。
浏览 1提问于2017-10-03得票数 12
回答已采纳
我希望实现一个从Kafka主题消费的函数，对于每个消费的消息，它调用另一个方法对消息进行一些处理，然后将其存储在数据库中。源kafka主题有12个分区。我想在同一个消费群体下创建6到12个kafka消费者。所有的消费者都将消费来自Kafka的消息并调用该函数
浏览 1提问于2020-02-19得票数 1
1回答
DLT消息消耗
、、、、
我们使用的是Spring kafka 2.7非阻塞重试机制。在Spring Kafka重试机制中，kafka监听器消费来自main topic、Retry Topic和DLT topic的消息，我们只希望监听器消费来自main和retry topic的消息。有没有一种简单的设置方法？
因为我们不希望同一个消费者
浏览 1提问于2021-10-14得票数 0
1回答
我看了一下stream 1.0.0.RELEASE的文档，似乎找不到任何关于错误处理的文档。
根据卡夫卡0.9的观察，如果我的消费者抛出一个RuntimeException，我会看到3次重试。<group>的消息，但是使用者偏移量没有被更新，滞后1。如果重新启动使用者，它尝试再次处理来自原始主题分区的相同失败消息，重试3次，并再次将相同的</e
浏览 0提问于2016-05-17得票数 4
回答已采纳
1回答
我们对Spring使用StreamListeners，并使用基于JPA的发件箱发送消息。发件箱是从消费中清空异步的，因此我们不希望在“从JPA表中读取，生成到Kafka的消息”上进行事务处理。但是我们也有重试主题，所以如果消费失败，我们将将失败的消息移动到重试主题(最后是DLT)，这需要是事务性的。
浏览 2提问于2022-03-30得票数 0
我有一个有很多分区的主题。我希望许多应用程序读取来自这个主题的所有消息。一些申请经常阅读，另一些则在午夜阅读。我怎样才能在卡夫卡实现这一点？
浏览 0提问于2018-10-19得票数 0
回答已采纳
我正在使用reactor-kafka库和spring webflux SSE流消费来自kafka主题的数据。当来自主题的所有消息都被消费完时，我需要返回一个特殊的ServerSentEvent，即最大主题偏移量get等于订阅时从0偏移量开始消耗的当前偏移量。以便客户端知道在kafka topic中没有更多的消息
浏览 7提问于2021-06-28得票数 1
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
