我的服务使用manual_immediate确认模式，自动偏移量重置为false。我在主代码中使用acknowledge.acknowledge()提交批处理记录，但在批处理错误处理程序中，我无法提交无效消息的偏移量。Issue1:需要了解确认批/提交偏移量的过程。Issue2:我正在以null形式获取数据。我尝试从data (为null)或thrownexception中读取原始消息，但失败。有没有人可以帮我处理一下<em
浏览 45提问于2020-08-09得票数 2
我正在实现spring批处理侦听器，它读取kafka主题的消息列表并将数据发布到REST服务。我想了解在REST服务下降时的偏移管理，批处理的偏移量不应该提交，下一次轮询的消息应该被处理。我已经阅读了spring文档，但在理解侦听器错误处理程序和在批处理中寻求当前容器错误处理程序之间的区别方面存在混淆。我使用的是spring 2.0.0.M7及以下版本是我的代码。
浏览 2提问于2017-12-13得票数 0
我刚开始使用spring-kafka 2.6.4。factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH);} 现在，我想定义适当的错误处理程序，使使用者停留在失败的记录上，而不是轮询下一批处理。我应该使用哪个错误处理程序？ 谢谢。
浏览 71提问于2021-01-19得票数 1
回答已采纳
我正在读Spring documentation，关于在kafka批处理消费中的错误处理。文档建议这样设置自定义错误处理程序： @Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer希望这对Spring社区有所帮助。
浏览 138提问于2019-05-03得票数 0
回答已采纳
我正在尝试使用启用批处理模式的spring云流来实现DLQ。return new RecoveringBatchErrorHandler(recoverer, new FixedBackOff(1000, 1));但是有几个问题：批处理中的存在多条消息，但如果第一条消息失败，则转到DLQ，但看不到下一条消息的处理。
浏览 11提问于2022-01-09得票数 0
回答已采纳
1回答
在我开始使用之前，我使用了Spring及其对批处理和自定义错误处理的支持。我只能找到以下配置属性：
因此，在中注册自定义错误处理程序并将AckMode设置为批处理是否(是否可行)？
浏览 4提问于2017-03-29得票数 0
回答已采纳
如何处理批处理模式下的反序列化异常？ }这里是我的应用程序属性:
spring.kafka.consumer.bootst
浏览 2提问于2021-04-08得票数 0
Situation
我正在尝试为(2.3.3.RELEASE) 批处理消息侦听器实现自定义错误处理程序。我使用的是“消费者感知”@KafkaListener，我的自定义错误处理程序也实现了ConsumerAwareBatchErrorHandler。在我的错误处理程序中，我使用Consumer.seek(topicPartition, offset)方法查找next或寻求当前偏移量。问题是，每当调用错误处理程
浏览 4提问于2019-12-06得票数 0
回答已采纳
1回答
我不完全理解提交偏移量和akcMode时消费者错误处理是如何工作的，以及错误时停止容器(使用Spring-Kafka1.3.*)是如何影响它的。第一个消费者--第一个事件被处理好，处理第二个事件失败--所以在错误处理程序中，我调用了kafkaListenerEndpointRegistry.stop()，但是由于stop是实现的，所以它只是停止消费者的轮询因此，第一个使用者处理事件3,4,5 (所有这些事件都没有错误处理)，并且假设第
浏览 3提问于2018-07-09得票数 3
回答已采纳
当我们将spring kafka升级到2.7.8时，我们在setAckOnError(false)方法中得到了错误，因为它已经被弃用了。现在有没有办法将错误的确认设置为false？还有没有其他方法可以帮助我将其设置为false以进行错误确认？附言:我是卡夫卡的新手，感谢大家的帮助！
浏览 221提问于2021-11-16得票数 0
回答已采纳
我使用和Spring函数创建了一个Kafka消费者，用于批量消费来自Kafka主题的消息。现在，我希望将错误批次发送到死信队列中，以便进一步调试错误。我正在用Spring重试处理我的使用者方法中的重试。但是对于不可重试的异常，我希望将整个批处理发送到DLQ。DeadRecordPublishinRecoverer启用错误处理程序将失败的消息发送到DLQ：public DefaultErrorHandler errorHa
浏览 9提问于2022-07-11得票数 0
我需要识别数据库连接失败异常与其他异常，并在我的kafka侦听器错误处理程序中进行相应的处理。由于Spring Data JPA中的所有内容都被转换为DataAcccessException，我只是想知道当应用程序运行时发生连接故障时会抛出什么异常。工作环境: Spring boot 2.x、Spring data jpa、spring-kafka、Java 8
浏览 2提问于2019-02-20得票数 1
我正在使用spring-kafka 2.2.8，试图理解单个记录消费者和批量消费者之间的主要区别是什么。 据我所知，从一个主题中读取消息/字节对于单个记录消费者和批量消费者来说没有什么不同。唯一的区别是如何提交偏移量。并因此进行错误处理。我的理解正确吗？请确认。
浏览 170提问于2020-10-17得票数 0
回答已采纳
我目前正在开发基于kafka的消息总线，由camel和Spring管理。我尝试了几种方法，首先将我的org.apache.camel:camel-kafka版本更新到2.22.0，以便能够进行手动提交。然后，我尝试实现一个基本的错误处理程序(配置了maximumRedeliveries=-1以进行无限重试)，这样当pollEnrich触发onException时，我可以设置一个头来避免执行最后的手动提交</e
浏览 1提问于2018-08-14得票数 4
问题陈述：@Bean(name ="kafkaConsumerFactoryConcurrentMessageListenerContainer<>(
} kafkaConsumerFactor
浏览 2提问于2021-01-06得票数 1
我正在为团队在他们的项目中使用构建一个通用的spring-kafka配置。 我想在容器级别定义一个通用的自定义错误处理程序，并允许项目为每个侦听器定义一个侦听器错误处理程序。任何未由侦听器错误处理程序处理的内容都应回退到容器。 根据我到目前为止的测试，它要么是一个，要么是另一个。有没有办法让他们一起工作？在容器级别拥有一个处理程序链并允许项目将错误处理程序添加到链中是否有意义？
浏览 10提问于2019-03-05得票数 0
1回答
批处理模式下的错误处理
、、、、
我正在使用和Kafka来批量使用来自Kafka主题的消息。我正在尝试实现一个错误处理机制。根据我的理解，我不能在批处理模式下使用Stream的enableDLQ属性。我发现RecoveringBatchErrorHandler和DeadLetterPublishingRecoverer可以重试并发送spring文档中的失败消息。我看到的所有例子都是使用KafkaTemplates。这就是我所指的春
浏览 5提问于2022-07-06得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
腾讯会议活动推荐
腾讯云开发者
