我指的是这个答案： https://stackoverflow.com/questions/56728833/seektocurrenterrorhandler-deadletterpublishingrecoverer-is-not-handling-deseria我们是否可以像下面这样添加手动即时确认： @KafkaListener(id = "so56728833", topics = "so56728833")
public
浏览 432提问于2021-06-14得票数 0
回答已采纳
Spring Kafka - How to Retry with @KafkaListener spring-kafka SeekToCurrentErrorHandler如果设置为spring.kafka.listener.ack-mode或者在指定的确认模式下重试工作，例如手动MANUAL_IMMEDIATE。@Value("${${sync.kafka.header.source.id}}"
浏览 78提问于2019-02-25得票数 0
回答已采纳
1回答
我有一个使用和Spring的应用程序，它处理Avro消息。应用程序工作正常，但现在我想添加一些错误处理。的目标：I希望捕获反序列化异常，构建一个带有异常细节的新对象+原始的Kafka消息+自定义上下文信息，并将该对象推送到一个专门的Kafka主题。基本上是一个DLQ，但原始消息将被拦截和修饰。MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON) }
以及属性文件：
浏览 1提问于2021-05-21得票数 2
我使用的是spring boot 2.1.7.RELEASE和spring-kafka 2.2.8.RELEASE，我们正在升级spring boot版本，但目前我们使用的是spring-kafka版本我正在使用@KafkaListener注解来创建一个消费者，并且我正在使用的consumer.And的所有默认设置都是在Spring-Kafka文档中指定的。来捕获发送导致反序列化的记录异常并将它们发送到DLT，从而实现了我的自定义<e
浏览 49提问于2021-06-26得票数 0
回答已采纳
我的要求是，当应用程序无法处理从通过spring boot java应用程序使用的kafka主题的当前偏移量读取的消息时，重置kafka主题的偏移量。在手动重置偏移量或发送否定确认之后，需要通过spring boot java应用程序的kafka消费者再次从不合适的偏移量值轮询消息。它能实现吗？
浏览 3提问于2019-01-09得票数 1
我正在使用spring boot 2.1.7.RELEASE和spring-kafka 2.2.8.RELEASE.And，我正在使用@KafkaListener注解来创建消费者，并且我正在使用的consumer.And的所有默认设置都是在Spring-Kafka文档中指定的。实现了我的自定义SeekToCurrentErrorHandler，但是记录值是null，并且记录键的格式不可读。请建议我如何获取记录键和记录值？下面是我的自定义Seek
浏览 0提问于2020-10-30得票数 0
1回答
spring.kafka.consumer.properties.spring.deserializer.value.delegate.class=io.confluent.kafka.serializers.KafkaAvroDeserializerspring.kafka.consumer.properties.spring.deserializer.key.delegate.class=org.apache.kafka.common.serial
浏览 21提问于2021-01-07得票数 0
回答已采纳
当我们将spring kafka升级到2.7.8时，我们在setAckOnError(false)方法中得到了错误，因为它已经被弃用了。现在有没有办法将错误的确认设置为false？还有没有其他方法可以帮助我将其设置为false以进行错误确认？附言:我是卡夫卡的新手，感谢大家的帮助！
浏览 221提问于2021-11-16得票数 0
回答已采纳
在春天的时候，
kafka: listener: retries: 3java web(bean) based configuration中使用spring boot，这是一个很有价值的优势。中的卡夫卡错误处理程序和</em
浏览 3提问于2021-12-03得票数 1
3回答
我已经为该功能设置了SeektoCurrentErrorHandler和DeadLetterPublishingRecoverer。at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:570) ~[spring-kafka-2.5.2.RELEASE.jar(FailedRecordProcessor.java:167) ~[spring-kafka-2.5.2.RELEAS
浏览 1提问于2020-08-07得票数 6
回答已采纳
在使用SeekToCurrentErrorHandler之后，我正在寻找一个非阻塞的kafka ErrorHandler。由于某些子系统不稳定，我们需要将高间隔时间设置为5分钟或更长时间。但是具有两个附加的报头值kafka_try-counter和kafka_try-timestamp。在SeekToCurrentErrorHandler和DeadLetterPublishingRecoverer的基础上，实现了RePublishingErrorHandler和
浏览 1提问于2020-03-11得票数 0
我们正在尝试将我们的Spring Cloud Stream应用程序从Kafka迁移到AWS Kenesis。我们需要手动确认来处理某些超时情况。对于Kafka，我们使用属性autocommitoffset设置为false，并使用确认头来处理手动确认。我看了一下Spring Cloud Stream的文档，看了下面的：
但是找不到任何解决方案。任何指针都会非常有帮助。
浏览 31提问于2021-03-05得票数 0
回答已采纳
我试图使用手动确认从一个主题消费多个消息，但如果所有消息都只有一次确认，则确认工作。}", concurrency = "${kafka.buyers.concurrency}"publicvoid listen( List<String> message, Acknowledgment ack ){}
在上面的代码中，如果我在<
浏览 4提问于2019-08-01得票数 1
我在我的Kafka消费者上面临着一些奇怪的重新平衡问题。我面临的问题是，当一个新的消费者被添加时，它会重新平衡，旧的消费者离开组，只有新的消费者接收和处理消息。我的消费者配置如下： kafka: ......thrownException, records, consumer, container); 我的应用程序处理每个事件大约需要10秒，并且只有在成功处理之后才会为每个事件发送确认kafka</
浏览 0提问于2021-01-15得票数 0
1回答
春云流卡夫卡粘结剂重试
、、、
我有春季云流应用与卡夫卡绑定，消费和发送信息。在应用程序中，我使用重试策略配置自定义错误处理程序，并向处理程序中添加不可还原的异常。>> customizer() { container.setErrorHandler(customErrorHandler);}
public S
浏览 6提问于2022-05-25得票数 0
回答已采纳
1回答
我使用的是spring-kafka 2.2.6。我用过SeekToCurrentErrorHandler和ErrorHandlingDeserializer2。SeekToCurrentErrorHandler当前配置为在三次重试后记录消息。有没有办法跳过验证错误(由Spring中的Validator实现捕获)和消息转换错误的重试？this.kafkaConfigProperties.getConsumerConcurrency());
factory.setA
浏览 29提问于2020-02-12得票数 0
回答已采纳
-kafka-tx- ERROR 1.TEST_TEST_XPTO.0错误org.apache.kafka.clients.Metadata -生产者org.apache.kafka.clients.MetadatatransactionalId#0-0-C-1 ERROR o.s.k.s.LoggingProducerListener -当发送带有key='null‘和有效载荷=’XPTOEvent(super=Event(id=CAPBA2548__，例如ABC_TEST_XPTO，所以我想用_ERR设置死信
浏览 7提问于2020-08-05得票数 0
我用的是弹簧引导2.1.9和弹簧Kafka 2.2.9 private String groupId;
@Value("${kafka.retry.maxAttem
浏览 4提问于2019-10-30得票数 0
回答已采纳
1回答
这是我的代码和配置。消费者代码： topics = "${spring.kafka.template.default-topic}", public void receive(Lis
浏览 7提问于2022-03-30得票数 0
回答已采纳
我写了一个卡夫卡消费者使用spring库(spring启动器-家长2.3.4.RELEASE)。我的代码中有下面的使用者配置 * configuration for kafka consumers at container level * @returnContainerProperties.AckMode.MANUAL_IMMEDIATE); }
* default kafka</em
浏览 23提问于2022-06-05得票数 2
回答已采纳
云服务器
ICP备案
云直播
对象存储
实时音视频
腾讯云开发者
