使用Spring cloud Stream Kafka监听器从kafka主题中读取消息，并在异常情况下将消息发送到死信队列，配置属性 spring.cloud.stream.kafka.bindings.input.consumer.enable-dlq=true
spring.cloud.stream.kafka.bindings.input.consumer.dlq-name=
浏览 15提问于2020-01-26得票数 1
1回答
我已经用kafka活页夹设置了一个春天的云流。正如前面提到的，我还启用了DLQ并设置了headerMode来捕获异常消息。然而，当有异常时，我只在我的kafka主题中看到消息，而没有异常详细信息。我的application.yml看起来像这样 application: cloud: defaultBinder: kafk
浏览 4提问于2017-10-13得票数 1
我使用的是卡夫卡绑定的春云流。使用spring.cloud.stream.kafka.bindings.function配置配置使用者，并启用DLQ。spring.cloud.stream.kafka.bindings.consumeTest-in-0: batch-mode: falsemaxAttempts: 5 backOffMax
浏览 9提问于2021-11-13得票数 1
回答已采纳
在这样做的同时，我们查看了kafka分区保留的磁盘空间。现在，错误主题使用与原始主题相同数量的分区。我们实际上并不需要分区的错误主题，kafka为每个分区分配了大量的空间。因此，问题是，是否可以使用最近引入的配置dlqProducerProperties为dlq设置较低的分区计数？当我们在代码中添加一些较低的分区计数时，我们面临着一些问题，即消息无法移动到dlq，因为目标分区不匹配。我猜测源消息的分区索引被重用，以将其移动到dlq目标(到相同的分区索引)，但它只有1个分区，而不是9个分区
浏览 15提问于2019-03-26得票数 0
回答已采纳
1回答
我试图为流创建一个自定义异常处理程序，以路由一些要请求的错误，另一些则是DLQ。在流应用程序的启动日志中，该bean看起来是正确的，并注册为errorChannel的侦听器，但出于某种原因，当抛出异常时，它们不会在我们的路由器中由onError云流和kafka绑定器配置使用所有默认设置
浏览 0提问于2018-11-27得票数 0
回答已采纳
1回答
我正致力于为银行创建一个大容量的JSON数据处理应用程序，使用Spring、Kafka和QuickFIX/J。这是我第一次使用Kafka和QuickFIX/J这样的技术，无法决定是使用普通的Kafka spring还是Kafka流还是spring云流。以下是要求：
从多个进程读取数据，并将数据发送给QuickFIX/J启动器，后者进一
浏览 7提问于2020-06-01得票数 0
我使用和Spring函数创建了一个Kafka消费者，用于批量消费来自Kafka主题的消息。现在，我希望将错误批次发送到死信队列中，以便进一步调试错误。我正在用Spring重试处理我的使用者方法中的重试。但是对于不可重试的异常，我希望将整个批处理发送到DLQ。TopicPartition("error.topic.name", 0)),}
但是这并不是向error.topic发送任何消息，从</
浏览 9提问于2022-07-11得票数 0
我们让多个应用程序使用者收听相同的kafka主题，并且生产者在向主题发送消息时设置消息头，以便特定实例能够评估标头并处理消息。request received message is {}",message.getName());在SpringCloudStream3.0.0中，@StreamListener被废弃了，我无法在函数中找到与条件属性等价的内容
浏览 11提问于2021-09-16得票数 5
我在spring boot应用程序的消费者接收器上使用@SendTo注释，没有参数。收到的消息包含头部"Kafka_replyTopic"，但@SendTo无法读取，kafka在发送回复时抛出异常： Listener failed; nested exception is java.lang.IllegalStateException
浏览 64提问于2020-09-02得票数 0
回答已采纳
我正在为事件驱动的应用程序开发spring云流和kafka绑定器(spring云流注释正在使用)
我能够在用户端使用死信队列处理异常，但无法在生产者端找到处理持久异常的相关文档。，即在发送消息时如何处理序列化/Size太大类型的异常(持久异常)。如果在发送消息时出现错误，并且无法恢复，我想要处理。有没有任何机制。
浏览 10提问于2022-09-15得票数 0
有4个主题：处理主题、异常主题、重试主题和拒绝主题。我有一个使用Kstream的处理器的spring云流应用程序。此处理器从异常主题读取消息，并基于每个消息中可用的标志创建两个kstream分支，用于重试主题和拒绝主题。现在需要做的是，在retry主题中出现的任何消息都必须等待一段特定的时间才能将其推回处理主题。谁能帮助我在春季云流中的kafka流应用程序中进行最好的设计或解决方案？能用磁通和Mon
浏览 2提问于2021-06-21得票数 0
1回答
的目标：I希望捕获反序列化异常，构建一个带有异常细节的新对象+原始的Kafka消息+自定义上下文信息，并将该对象推送到一个专门的Kafka主题。基本上是一个DLQ，但原始消息将被拦截和修饰。问题：虽然可以拦截异常，但我不知道如何从Kafka获取原始消息(TODO 1，下面)。: org.apache.kafka.common.serialization.StringDeserializer
spring
浏览 1提问于2021-05-21得票数 2
1回答
在使用ErrorHandlingDeserializer结合Avro处理错误时，我无法发布到Dlq主题。下面是发布时的错误。
这是appli
浏览 3提问于2022-03-02得票数 0
回答已采纳
1回答
我有以下集群：我的问题是如何选择最有效的日志收集器(或者其他一些允许管理Kafka和ElasticSearch之间数据流的软件)。我正在尝试从Logstash、Fluentd和Confluent的Kafka Elasticsearch连接器中进行选择。我面临的主要问题是无法在写入Elasticsearch端点时在Kafka中回滚偏移。
例如，logstash说“如果启用400和404错误将发送
浏览 1提问于2018-10-19得票数 0
我们通过集成使用Kafka流。我将复制因子配置为在所有内部Kafka流主题中使用，方法是设置它适用于大多数由Kafka流在内部使用的重新分区/变更主题。这是卡夫卡流中已知的臭虫吗？我什么也找不到。如果是的话，是否有办法提高这些变更主题的复制因子？
我们在代理端使用<
浏览 5提问于2020-10-14得票数 1
回答已采纳
1回答
我的用法是接收卡夫卡消息，多次尝试一个rest呼叫，并在耗尽时将失败的消息转储到Kafka主题上。发布(enableDlq: true)，异常需要泡沫化到@StreamListener注释的方法，以便卡夫卡绑定器完成所需的工作。但是，在这样做时，我无法利用@Recover注释的方法，在该方法中，流在重试后完全落地：public boolean recoverOnToDLQ(OneException ex,String message, String throw ex; /&#x
浏览 2提问于2018-07-20得票数 0
我正在开发一个Spring Boot应用程序，它使用以Kafka主题为源的Spring集成流。我们的集成流开始使用一个包含带有springframework.cloud.stream.annotation.Input和输出注释的SubscribableChannels的接口。这些配置为通过云配置和spring.cloud.stream.kafka.bindings从Kafka读取。
当应用程序第一次
浏览 0提问于2018-04-24得票数 1
1回答
春云流不抛异常
、、
我正在使用春云流与rabbitMQ。我试图通过删除运行时的队列并将消息发送到已删除的队列来进行负面测试。我原以为上面的代码会抛出一个异常，但事实并非如此。甚至从队列1读取的消息也已被确认。队列1和队列2上有一个死信队列，但是消息没有进入dlq。
浏览 0提问于2018-07-05得票数 0
2回答
我希望只在消息处理成功时使用spring云流手动提交偏移量。这是我的代码- application.yml & Handler类 }--------------------------------------------------------------------------------- application: cloud:
浏览 3提问于2021-02-08得票数 2
回答已采纳
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
