当并发级别小于分区数时，Spring Kafka SeekToCurrentErrorHandler maxFailures无法工作

文章/答案/技术大牛

发布

2回答

apache-kafka、spring-kafka

我使用的是spring kafka 2.2.7，我的消费者配置代码如下： @Slf4j@EnableKafka throw new RuntimeException("force to retry");} 因此，如果我的侦听器中发生了异常，即使我在配置中配置了maxFailures，如果我指定的并发级别小于我的主题的<em

浏览 76提问于2019-09-11得票数 0

1回答

SeekToCurrentErrorHandler: DeadLetterPublishingRecoverer未处理反序列化错误

spring-kafka

我正在尝试使用spring-kafka版本2.3.0.M2库编写kafka consumer。只有当我的消费者代码抛出异常时，它才能正常工作，但当无法反序列化消息时，它就会失败。下面是我的kafka属性 spring: consumer: group-id: org.ap

浏览 56提问于2019-06-24得票数 3

回答已采纳

1回答

分区存在时，ConcurrentMessageListenerContainer正在减少并发数

java、spring、spring-boot、apache-kafka

我有一个spring boot应用程序，用于生成kafka的消息。此应用程序在高流量下运行，每天在6个实例上执行1000万次请求。我有一个弹簧启动卡夫卡消费应用程序也。但此应用程序有2个实例，这些实例无法使用所有消息，因为此应用程序正在运行单线程。我的主题有4个分区，我想根据分区数做消费者应用程序多线程。但我不确定我的代码是否能正常工作。Spring Kafka maven依赖 <g

浏览 17提问于2019-11-29得票数 0

回答已采纳

2回答

spring中的并行处理与自动缩放

spring-boot、apache-kafka、kafka-consumer-api、spring-kafka、autoscaling

我正在使用spring来使用来自两个Kafka主题的消息，它发送的消息格式如下所示。如果主题中的消息数量更多，我需要自动调整我的微服务来启动新的实例(直到分区的数量)。从KafkaListener的角度来看，我可以依赖哪些参数(CPU、内存)来查找主题中的消息数量更高？

浏览 1提问于2020-05-24得票数 2

回答已采纳

1回答

给定时间段的批处理消费者不能在Stream(StreamListener)中处理多个分区？

apache-kafka、batch-processing、spring-cloud-stream

: stream: binders: type: kafkaenvironment: cloud: kafka在StreamListener中，当分区计数为1时，我可以在每5000毫秒内以批处理方式使

浏览 5提问于2020-11-16得票数 1

回答已采纳

2回答

无法找到合适的方法来应对卡夫卡春天的复苏

java、spring-boot、spring-kafka

我正在开发一个相当简单的Spring应用程序，它将处理来自单个主题的消息，然后为每个消息调用一个外部web服务。我希望此服务在错误方面具有一定的智能性，因此，如果外部网站在短时间内无法访问，则应该重新尝试记录，并进行曝光退避，直到我们放弃并只记录错误并提交记录的某个点。我正在使用Spring和SpringKafka2.3.3。我将向您展示一下我为这个设置提供的Spring配置。有些事情是为了简洁而遗漏的。询问是否有一些价值或其他配置可能有帮助。eh = new SeekToCurrentErrorHandler<

浏览 1提问于2019-12-16得票数 1

回答已采纳

1回答

弹簧批处理-员工崩溃时重新启动行为

spring、spring-boot、apache-kafka、spring-batch

当使用远程分区时，我一直在探索Spring批处理在某些失败情况下如何工作。所有节点都已开始处理，但突然有一个节点崩溃。已崩溃的节点将

浏览 5提问于2022-03-31得票数 1

回答已采纳

1回答

用@KafkaListener和监听器的ack模式设置为记录的Spring* Kafka轮询*

apache-kafka、kafka-consumer-api、spring-kafka

我使用@KafkaListener和ConcurrentKafkaListenerContainerFactory来听3个kafka主题，每个主题有10个分区。listener.ackmode返回，enable.auto.commit设置为false和partition.assignment.strategy: partition.assignment.strategy 1)我对并发性的理解是，因为我将并发性(在工厂级别)设置为30，并且总共有30个分区(所有三个主题一起阅读)，每个线程

浏览 5提问于2019-11-03得票数 4

回答已采纳

1回答

什么应该是卡夫卡消费者并发的适当价值(关于比例)？

apache-kafka、spring-kafka

这是Spring应用程序，所以我使用spring。当只有两个服务实例时，每个实例运行10个线程，每个线程从一个分区读取。据我所知，当一个新的消费者加入一个消费者组时，一组消费者试图“重新平衡”负载，以便为每个消费者分配<e

浏览 0提问于2019-02-06得票数 2

1回答

seekToCurrentErrorHandler -kafka-当有来自不同分区的多个失败记录时，spring会随机运行，消费者会离开代理

spring-boot、spring-kafka

我们在我们的项目中使用spring-kafka version2.2.6-RELEASE(带有spring-boot)。看到(并面临问题)，在单个使用者(和并发1)具有来自不同分区的多个失败记录的情况下，它最终在无限循环( https://github.com/spring-projects/spring-kafka/issues当一个kafka消费者从一个主题(有3

浏览 274提问于2020-08-11得票数 2

1回答

如何使用属性在功能性出站通道中设置消息的分区？

spring-cloud-stream、spring-cloud-stream-binder-kafka、spring-cloud-stream-binder

我有一个非常简单的应用程序，它使用函数式方法来使用来自Kafka主题的消息，处理它们的有效负载并将它们发送到另一个主题。我正在使用Spring 2.7.5和3.2.6 (由Spring 2021.0.5管理)。入站消息包含一个头MY_PARTITION_ID，它是一个整数，指示应该生成出站消息的分区的数目。但是，当启用partitionKeyExpression时，我希望看到所有输出消息都转到输出主题的分区3。尽管如此，它们都进入了分区0。它正在<e

浏览 10提问于2022-11-19得票数 2

1回答

Flink Kafka* -如何使应用程序并行运行？*

java、parallel-processing、apache-kafka、apache-flink

我正在用Flink创建一个应用程序我该怎么做？than 1) properties.setProperty("auto.offset.reset&quo

浏览 1提问于2017-09-21得票数 2

回答已采纳

1回答

春云流产生与消费的主题

spring-boot、spring-cloud-stream

当我第一次启动服务并且这个主题在Kafka中不存在时，会引发以下异常： java.lang.IllegalStateException: The number of expected partitions(KafkaMessageChannelBinder.java:149) [spring-cloud-stream-binder-kafka-1.2.1.RELEASE.jar!(KafkaMessageChannelBinder.java:88) [spring-cloud-stream-binder-

浏览 5提问于2017-09-28得票数 1

1回答

春季-卡夫卡(非整合)消费者不消费信息

spring、apache-kafka、spring-kafka

我正在将我的应用程序集成到spring(不是spring-integration-kafka)。以下是项目的spring文档：@EnableKafka partition.a

浏览 3提问于2016-07-08得票数 3

1回答

如果使用的是不同的线程，则使用者dlq和errorChannel不能工作。

java、spring-kafka、spring-cloud-stream、spring-cloud-stream-binder-kafka

这种方法的问题在于，工作需要在一个单独的线程中管理，任何抛出的异常最终都不会在errorChannel和DLQ中结束。(cr, e) -> new TopicPartition("ERR", 1)), } enable-dlq没有在spring.cloud.stream.kafka.bindings.channel-name.consumer虽然我们需要进行一些修改，因为我们使用了不推荐的StreamListner方法而不是函数，但是有一些问

浏览 1提问于2021-04-28得票数 3

回答已采纳

1回答

应用程序无法启动，因为kafka* streams存储正在等待运行*

java、spring-boot、apache-kafka、apache-kafka-streams

我有一个使用kafka streams的spring boot应用程序(kafka docker镜像：wurstmeister/kafka:2.12-2.1.1，kafka依赖：org.apache.kafka当新版本的应用程序准备就绪时，k8s会启动新应用程序并缩小旧应用程序的规模。问题是，当新的应用程序启动时，我在日志中只看到多行，比如："Waiting for store my-topic-

浏览 1提问于2020-05-22得票数 0

1回答

卡夫卡再平衡澄清

apache-kafka、spring-kafka

我使用spring 1.3.5，它使用0.11.0.2客户端。我有两个分区。我的max.poll.records = 1。我的问题是:假设我有2个节点/工作人员(其并发性设置为一个，因此每个节点中有一个使用者)是慢速工作人员(处理记录可能需要5分钟)，这两个节点当前都处于关闭状态。当节点-1(使用者-1)加入消费者组时，它得到分配给它的两个分区(p-1，p-2)。现在，使用者-1消耗来自p-1的消息。节点2(消费者-2)立即加入相同的消费者组

浏览 1提问于2018-05-24得票数 1

2回答

将消息从Kafka路由到连接到应用服务器集群的web套接字客户端

websocket、apache-kafka、spring-kafka

我理解spring为使用和发布到kafka主题的消息提供了便利，但是在负载平衡的应用服务器场景中，当连接到分布式kafka主题时，这是如何工作的。消息保存在数据库中，并发布到一个kafka主题/分区中，相应的web客户/用户可以使用该主题/分区。卡夫卡可以扩大到许多经纪人与许多消费者。我可以看到如何在一个应用服务器中很容易地实现这一点，在这个服务器中，使用者使用来自一个ka

浏览 4提问于2019-10-15得票数 5

回答已采纳

3回答

KafKa分区器类，使用键将消息分配给主题内的分区。

apache-kafka

每个主题被划分为一个或多个分区(每个分区可以有多个副本)。所以他们的行为是平行的在卡夫卡0.7 中，他们说 ProducerData<String, String>

浏览 1提问于2013-08-13得票数 31

13回答

60000 ms后元数据中没有出现org.apache.kafka.common.errors.TimeoutException:主题

java、windows、apache-kafka、kafka-producer-api

我得到了错误：import org.apache.kafka<

浏览 17提问于2020-09-02得票数 19

回答已采纳

点击加载更多