kafka重新平衡过程中的Poll call是否为忙碌等待？
在kafka重新平衡过程中，Poll call是一种忙碌等待的机制。当kafka集群中的消费者发生变化（例如新的消费者加入或者已有的消费者退出），会触发重新平衡过程。在重新平衡期间，kafka会暂停消费者的消息消费，并重新分配分区给消费者，以保证消费者组内的负载均衡。
在重新平衡过程中，消费者会调用Poll方法来获取新的分区分配信息。如果消费者在重新平衡期间调用Poll方法，但没有获取到新的分区分配信息，它会进入忙碌等待状态。在这种状态下，消费者会不断地发送Poll请求给kafka集群，直到获取到新的分区分配信息为止。
忙碌等待的机制可以确保消费者在重新平衡期间不会错过任何消息。当重新平衡完成后，消费者会收到新的分区分配信息，并继续消费消息。
需要注意的是，由于kafka的重新平衡过程可能涉及到网络通信和分区分配计算等耗时操作，因此在重新平衡期间，消费者的消息消费可能会暂时中断。为了提高消费者的可用性和性能，可以通过合理设置消费者的参数和调整kafka集群的配置来优化重新平衡的速度和效率。
推荐的腾讯云相关产品：腾讯云消息队列 CKafka（https://cloud.tencent.com/product/ckafka）是腾讯云提供的高可用、高吞吐量的消息队列服务，基于kafka开源技术，适用于大数据实时计算、日志采集、消息通信等场景。它提供了稳定可靠的消息传输，支持自动分区和负载均衡，能够满足各种规模的应用需求。
相关·内容
我正在使用kafka-clients 2.5.1，并且有一个具有以下属性的kafka streams应用程序：
properties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, "1");
properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, String.valueOf(Integer.MAX_VALUE));
properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
prop
浏览 2提问于2021-07-02得票数 0
背景：I有一个在Kubernetes中运行的应用程序，它使用Kafka作为集中消息总线。我的应用程序中的Kafka客户端可能非常慢。Kafka在随后的轮询()调用之间等待的最长时间(在将成员踢出组和重新平衡之前)由max.poll.interval.ms控制。
对于这个应用程序中的大多数工作人员，我可以将max.poll.interval.ms设置为几分钟的顺序。然而，对于速度较慢的工人，我需要将其设置为几个小时。
当一切正常工作时，这不会引起问题。然而，在网络中断或间歇性崩溃的情况下，我注意到最大投票间隔很大的工作人员可能会“陷入”再平衡。如果我在发生这种情况时查看代理，并执行以下操作
ka
浏览 6提问于2020-04-15得票数 1
AFAIK，max.poll.interval.ms是在Kafka 0.10.1中介绍的。但是，当我们可以同时使用session.timeout.ms和max.poll.interval.ms时，还不清楚是否考虑使用can，其中心跳线程没有响应，但是我的处理线程具有较高的值集，它仍然在处理记录。但是当心跳线程下降之后，在穿越session.timeout.ms之后，到底会发生什么。因为我在POC中观察到，消费者再平衡在到达max.poll.interval.ms之前是不会发生的。
所以对我来说，session.timeout.ms似乎是多余的。
类似的发布了，但它没有回答这个问题。
浏览 4提问于2020-02-04得票数 2
回答已采纳
2回答
我已经阅读了其他帖子，我已经通过使用新的组ID绕过了这个问题，但是我想知道是什么导致了这个问题。
我有一个有16个分区的主题，我设置了session.timeout.ms=30000和max.poll.interval.ms=30000000。
我运行我的程序，然后ctrl+c它，所以它不能正常关闭。在我猜了16次之后，我陷入了这个重新加入的问题中。session.timeout.ms是心跳超时，所以在30秒后，它应该踢我的消费者权利，我的分区应该“释放”对吗？或者它只是在听我的max.poll.interval.ms？
编辑:我仍然间歇性地收到这个错误，当它发生时，我必须重新启动我的所有使用
浏览 4提问于2017-06-15得票数 13
使用从只有少数消息的主题中读取消息。卡夫卡一遍又一遍地重复队列中的信息。
它接收来自我的主题的一条消息(返回到每个消息内容)，然后抛出ERROR - Heartbeat session expired - marking coordinator dead并继续遍历其余的消息，并继续重播它们。更多的日志：
kafka.coordinator - ERROR - Heartbeat session expired - marking coordinator dead
kafka.coordinator - WARNING - Marking the coordinator dead (node 1
浏览 15提问于2017-09-29得票数 2
我对kafka消费者有问题，它有时会抛出异常。
ERROR [*KafkaConsumerWorker] (Thread-125) [] Kafka Consumer thread 235604751 Exception while polling Kafka.: org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another membe
浏览 2提问于2020-06-10得票数 0
我正在做一个卡夫卡消费者项目。最近我们在PROD环境中部署了它。在那里，我们遇到了如下问题：
[main] INFO com.cisco.kafka.consumer.RTRKafkaConsumer - No. of records fetched: 1
[kafka-coordinator-heartbeat-thread | otm-opl-group] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-otm-opl-group-1, gr
浏览 4提问于2020-03-30得票数 4
今天，在我的Spring和单个实例Kafka应用程序中，我面临以下问题：
无法完成org.apache.kafka.clients.consumer.CommitFailedException:提交，因为组已经重新平衡并将分区分配给另一个成员。这意味着随后调用轮询()之间的时间要比配置的max.poll.interval.ms长，这通常意味着轮询循环花费了太多的时间处理消息。您可以通过增加会话超时或减少max.poll.records在轮询()中返回的批处理的最大大小来解决这一问题。
原因可能是什么，以及如何解决？据我所知，我的消费者被封锁了很长一段时间，对心跳没有反应。我应该调整卡夫
浏览 0提问于2018-08-01得票数 7
回答已采纳
2回答
我有一个Kafka流应用程序，它从几个主题中获取数据，并将数据加入到另一个主题中。
Kafka配置：
5 kafka brokers
Kafka Topics - 15 partitions and 3 replication factor.
注:我在卡夫卡经纪人运行的同一台机器上运行卡夫卡流应用程序。
每小时消耗/产生数以百万计的记录。每当我把任何卡夫卡经纪人打倒，它就进入再平衡，它需要大约。30分钟，有时甚至更长的时间再平衡。
有人知道如何解决卡夫卡消费者的再平衡问题吗？而且，它多次在重新平衡的同时抛出异常。
这将阻止我们在生产环境中使用此设置。任何帮助都将不胜感激。
Caused by
浏览 3提问于2017-09-09得票数 6
回答已采纳
我试图设置使用者组属性，使其只能每分钟使用一条消息。
consumer.properties
group.id=test_group
max.poll.interval.ms=60000
session.timeout.ms=60000
max.poll.records=1
生产者
kafka-console-producer.sh --broker-list localhost:6667 --security-protocol SASL_PLAINTEXT --topic cdc_poc
Consumer
kafka-console-consumer.sh --bootstrap-serv
浏览 2提问于2019-03-01得票数 1
回答已采纳
1回答
在kafka使用者中，如果消息的处理时间超过5分钟，则消息是新处理的，我已经将使用者配置为增加"max.poll.interval.ms“和session.timeout.ms。
max.poll.interval.ms= 7200000 (2小时) session.timeout.ms= 7200000 (2小时) request.timeout.ms=7206000 (~2小时)
这开始起作用了，在那之后，它是重新平衡的，我在消费者身上得到了这个错误。
2019-10-18 15:29:51.739 INFO o.a.k.c.c.i.AbstractCoordinator -
浏览 3提问于2019-10-18得票数 4
1回答
使用kafka版本2.11-0.11.0.3发布10,000条消息(所有消息的总大小为10 as )，将有2个使用者(具有相同的组-id)作为并行处理来使用该消息。在消费过程中，同一条消息被两个使用者使用。
下面的错误/警告是由kafka抛出的
警告:此成员将离开组，因为使用者投票超时已过期。这意味着后续调用轮询()之间的时间要比配置的max.poll.interval.ms长，这通常意味着轮询循环花费了太多时间处理消息。您可以通过增加max.poll.interval.ms或减少在max.poll.records ()中返回的批处理的最大大小来解决这一问题。
信息:尝试心跳失败，因为小组正
浏览 0提问于2019-03-21得票数 0
1回答
我在这里询问的是关闭卡夫卡的消费者。即使使用线程退出，我也需要关闭kafka使用者吗？会不会因为任何改变而关闭它的资源？
下面是一个代码示例：
public class MyThread extends Thread{
private KafkaConsumer<String, Message> kafkaConsumer;
@Override
public void run() {
kafkaConsumer = initConsumer();
while(true){
kafkaconsume
浏览 0提问于2018-08-16得票数 3
回答已采纳
我用春天的卡夫卡来消费卡夫卡的信息。使用者侦听器如下所示。
@KafkaListener(topics = "topicName",
groupId = "groupId",
containerFactory = "kafkaListenerFactory")
public void onMessage(ConsumerRecord record) {
logger.info("Received Message from kafka topic " + reco
浏览 45提问于2022-05-27得票数 1
我们正在运行一个3代理Kafka 0.10.0.1集群。我们有一个java应用程序，它可以从不同的主题中产生许多消费者线程。对于每个主题，我们指定了不同的消费群体。
很多时候，我发现只要重启这个应用程序，一个或多个CG接收分区分配的时间就会超过5分钟。在此之前，该主题的消费者不会消费任何东西。如果我转到Kafka broker并运行consumer-groups.sh并描述那个特定的CG，我会看到它正在重新平衡。在server.log中，我看到这样的代码行
准备稳定组otp-sms-consumer稳定组otp-sms-consumer
这两个日志之间通常会有大约5分钟或更长的间隔。在消费者端
浏览 2提问于2018-07-13得票数 10
1回答
我正在使用Kafka 0.10.2，为了接收Kafka的记录，我有一个消费者投票循环，如下所示：
while (true) {
ConsumerRecords<Long, String> records = consumer.poll(2000);
int count= 0;
long lasttimestamp = 0;
long lastOffset = 0;
for (ConsumerRecord<Long, String> record : records) {
if (record.timestamp()
浏览 2提问于2017-10-08得票数 0
在我的kafka streams应用程序中创建状态存储时，我得到了这个错误Failed to lock the state directory: /tmp/kafka-streams/string-monitor/0_1。下面是应用程序的完整堆栈跟踪
[2016-08-30 12:43:09,408] ERROR [StreamThread-1] User provided listener org.apache.kafka.streams.processor.internals.StreamThread$1 for group string-monitor failed on partiti
浏览 19提问于2016-08-30得票数 4
2回答
我使用的是卡夫卡连接器汇合3.0.1版本，我创建了一个名为 new - group 的新组，it.most上大约有20个主题是busy.But。遗憾的是，当我启动连接器框架时，系统无法停止再平衡，所有主题的再平衡时间大约为2分钟。我不知道原因。其中一些错误消息是：
[2017-01-03 21:43:57,718] ERROR Commit of WorkerSinkTask{id=new-connector-0} offsets threw an unexpected exception: (org.apache.kafka.connect.runtime.WorkerSinkTask:1
浏览 0提问于2017-01-03得票数 7
问题
我们使用StatefulSet在Kubernetes上部署Streams应用程序。实例有不同的applicationId，因此它们都复制了完整的容错输入主题。它们本质上是只读服务，只在状态主题中读取并将其写入状态存储，在那里通过REST服务客户请求。这意味着，在任何给定的时间，消费者群体总是只包含一个Kafka流实例。
我们现在的问题是，当触发滚动重新启动时，每个实例大约需要5分钟才能启动，其中大部分时间是在REBALANCING状态下等待的。我读过，Kafka不发送LeaveGroup请求，以便在容器重新启动后快速返回，而不进行再平衡。为什么这对我们不起作用，为什么再平衡需要这么长时间
浏览 0提问于2020-05-08得票数 5
回答已采纳
我使用手动kafka commit，将属性enable.auto.commit设置为false，同时初始化Kafka消费者，并在接收和处理消息后手动调用kafka commit。
然而，因为在我的消费者中处理消息是很耗时的，所以我得到了Exception with message "error": "Broker: Group rebalance in progress"
原因是重新平衡超时后提交被拒绝，并出现此错误。现在，恢复操作是退出并重新实例化进程，这将再次触发重新平衡和分区分配。另一种方法是捕获此异常，然后照常继续，只有在poll()调用被阻塞直到重新平
浏览 15提问于2020-05-19得票数 0
我当时使用的是卡夫卡0.10.2，现在我面对的是CommitFailedException。像这样：
无法完成提交，因为组已经重新平衡并将分区分配给其他成员。这意味着随后调用轮询()之间的时间要比配置的max.poll.interval.ms长，这通常意味着轮询循环花费了太多的时间处理消息。您可以通过增加会话超时或减少max.poll.records在轮询()中返回的批处理的最大大小来解决这一问题。
我已将max.poll.interval.ms设置为Integer.MAX_VALUE。那么，有人能告诉我，为什么这种情况仍然发生，即使我已经设定了价值？
另一个问题是:我按照描述将ses
浏览 7提问于2017-08-08得票数 45
当kafka代理启动并运行时，在集群模式下运行的spark进程能够读取来自kafka主题的消息。但是当代理被故意关闭时，spark消费者仍然处于运行状态。
是否有任何参数来控制spark消费者和zookeeper进程之间的握手间隔，以便如果无法访问代理，spark进程可能会失败。或者有没有让消费者失望的替代方法。请提个建议。
浏览 0提问于2021-02-18得票数 0
1回答
在春天-卡夫卡，我想重新消费一个卡夫卡的主题从一开始。通过将group.id转换为卡夫卡所不知道的东西，当然可以做到这一点：
@KafkaListener(topics = "sensordata.t")
public void receiveMessage(String message) {
...
}
@Bean
public Map consumerConfigs() {
Map props = new HashMap<>();
props.put(ConsumerConfig.GROUP_ID_CONFIG, "NewGroupID
浏览 7提问于2016-11-14得票数 0
需要一些帮助，找出我收到的一个卡夫卡流消费者的例外。
我用低级别处理器API实现了Kafka流。对于从Kafka接收到的每一次更新，都会将其合并并更新到keystore，从而维护状态。最初，我们只运行了一个消费者，过了一段时间，我们尝试提出第二个消费者。但第二个消费者在再平衡期间抛出了一个例外，称其未能实现再平衡。这是因为更改日志的状态发生了更改(下面的异常共享)。我假设，当重新平衡发生时，第一个使用者收到了一些更新，因此更新被推送到相应的更改日志中。请帮帮忙。还共享相同的示例代码。我使用的是Kafka 2_11 0.10.2.1，主题有72个分区
异常
Exception in th
浏览 2提问于2017-08-30得票数 1
回答已采纳
1回答
卡夫卡增量粘性再平衡
、、、
我用卡夫卡·斯特里齐运算符在库伯奈特上运行卡夫卡。我使用增量粘性再平衡策略，使用以下方法配置我的消费者：
ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
org.apache.kafka.clients.consumer.CooperativeStickyAssignor.class.getName()
每次我缩放我的消费者组中的消费者时，该组中的所有现有消费者都会生成以下异常
线程“主”org.apache.kafka.common.errors.RebalanceInProgressException:偏移提交中的_Ex
浏览 1提问于2021-06-21得票数 4
我在python中创建了Kafka使用者，配置如下。
consumer = KafkaConsumer(topic,
group_id='consumer',
bootstrap_servers=[bootstrap_servers],
auto_offset_reset='latest',
value_deserializer=lambda m:
浏览 5提问于2022-06-15得票数 0
1回答
在分区0上测试一条Kafka消息，处理时间超过一个小时。我尝试过自动提交，暂停/恢复，增加最大轮询超时，但我无法停止消费者消息复制。这是使用python的融合式kafka。我看不出有什么方法可以实验异步/同步提交。
msg= self.consumer.poll(timeout=1.0)
if msg:
self.consumer.pause()
some task.. (takes ~60 minutes minutes)
self.consumer.res
浏览 3提问于2022-03-07得票数 0
我用春季卡夫卡2.7.8来消费卡夫卡的信息。使用者侦听器如下所示
@KafkaListener(topics = "topicName",
groupId = "groupId",
containerFactory = "kafkaListenerFactory")
public void onMessage(ConsumerRecord record) {
}
上面的onMessage方法一次接收单个消息。
这是否意味着春库将max.poll.records设置为1，还是每次轮询5
浏览 22提问于2022-05-19得票数 0
1回答
我在用
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.0.1</version>
</dependency>
下面的代码返回非空分配的分区，但poll(0)不推荐.
val records = kafkaConsumer.poll(0) // <= deprecated
logInfo(s"Dummy call
浏览 0提问于2019-04-25得票数 0
我正在尝试做一个简单的poc与spring kafka与spring boot与版本(2.3.7版本)，以实现如何消费者批处理工作，以及如何重新平衡工作，如果消费者需要更多的处理时间，因为我对这个消息传递系统完全陌生。 现在我看到了kafka与单一消费者(不允许并发)重新平衡的问题。 这些是我设置了max.poll.interval.ms = 50000和factory.getContanerProperties.setIdealTimeBetweenPolls(120000)的属性，并使消费者批处理比这个max.poll.interval.ms占用更多的时间。(70秒)我在这里看到的问题是
浏览 98提问于2021-02-01得票数 1
我已经开发了卡夫卡消费者应用程序使用spring-kafka库和使用默认消费者配置与手动提交。
我正在运行两个应用程序实例，侦听两个不同的Kafka主题。在执行负载测试时，我注意到只有一个用于较高负载的应用程序出现了以下错误：
Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member.
This means that the time between subsequent calls to poll() was lon
浏览 1提问于2018-07-26得票数 0
用我们的春季引导应用程序，我们注意到卡夫卡消费者消费信息随机两次，偶尔只在prod中使用。我们有6个实例，在PCF.We中部署了6个分区，捕获了具有相同偏移量的消息，并且在同一个主题中接收了两次分区，这会造成重复，对我们来说是业务上的关键。我们还没有注意到这一点，在非生产环境中，是很难复制在非生产env。我们最近转向卡夫卡，我们无法找到根本的问题。
我们使用的是spring-cloud-stream/spring-cloud-stream-binder-kafka- 2.1.2，这里是Config：
spring:
cloud:
stream:
default.cons
浏览 7提问于2022-02-07得票数 0
我们有一个具有以下配置的KStreams应用程序：
props.setProperty(RETRIES_CONFIG, String.valueOf(Integer.MAX_VALUE));
props.setProperty(RETRY_BACKOFF_MS_CONFIG, "5000"); // 5 seconds
props.setProperty(RECONNECT_BACKOFF_MS_CONFIG, "5000"); // 5 seconds
props.setProperty(REQUEST_TIMEOUT_MS_CONFIG, "5000
浏览 2提问于2020-02-07得票数 7
在创建多个使用者(使用Kafka0.9javaAPI)并启动每个线程之后，我将得到以下异常
Consumer has failed with exception: org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed due to group rebalance
class com.messagehub.consumer.Consumer is shutting down.
org.apache.kafka.clients.consumer.CommitFailedExceptio
浏览 0提问于2016-02-26得票数 8
我使用了spring-cloud-starter stream-kafka 3.0.4，我有以下错误：
org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than th
浏览 10提问于2022-06-27得票数 0
我正在使用带有recordFilterStrategy的spring-kafka。 @Bean("manualImmediateListenerContainerFactory")
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Object, Object>> manualImmediateListenerContainerFactory(
ConsumerFactory<Object, Object> cons
浏览 50提问于2021-09-20得票数 1
所以基本上我看到的是，我正在使用@KafkaListener消费消息，我故意不想承认它们，所以消费者忽略它们，并在主题中侦听新的消息。因此，如果应用程序重新启动，使用者只会从最后提交的消息开始。因此，我希望避免这种情况，我希望消费者在不重新启动的情况下考虑未提交的消息。下面是我正在使用的配置和@kafkalistener。
ReceiverConfig.java
package com.config;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serializ
浏览 3提问于2019-12-05得票数 1
2回答
Kafka Broker有3个分区。
卡夫卡消费者实例的计数是3。
突然，一个消费者的例子死了。
我知道，如果一个Kafka消费者实例死亡，Kafka代理正在进行再平衡，而另一个消费者实例将被分配到该分区。
我想知道这样的假设是否正确:另一个实例使用它最初使用的所有分区，然后分配和使用死分区。
(我是否必须在客户端代码中实现ConsumerRebalanceListener？)
如果是这样的话，在使用消息时会有什么延迟吗？
谢谢。
浏览 3提问于2017-05-17得票数 1
Apache Kafka文档指出：
内部Kafka流消费者max.poll.interval.ms默认值从300000改为Integer.MAX_VALUE。
由于此值用于检测一批记录的处理时间超过给定阈值的时间，是否存在这样一个“无限”值的原因？
它是否使应用程序变得没有响应能力？或者，卡夫卡流有一种不同的方式离开消费者群体时，处理过程太长？
浏览 1提问于2017-12-20得票数 12
回答已采纳
我在我的Kafka消费者上面临着一些奇怪的重新平衡问题。我使用SeekToCurrentErrorHandler实现了无限重试策略，将max.poll.interval.ms=1,200,000设置为20分钟，并将重试延迟设置为900,000 15分钟延迟。
如您所见，poll.interval > delay。我面临的问题是，当一个新的消费者被添加时，它会重新平衡，旧的消费者离开组，只有新的消费者接收和处理消息。在新的消费者中，我看到了日志Attempt to heartbeat failed since group is rebalancing，但它仍然在处理消息。旧的消费者接收0数
浏览 0提问于2021-01-15得票数 0
此接口- rd_kafka_poll_set_consumer()的用途是什么？在rdkafka.h文件中，它指出一旦您将rd_kafka_poll队列(rk->rep)重定向到rd_kafka_consumer_poll的队列(rkcg->rkcg_q)，在此之后您就不应该调用rd_kafka_poll。
谁能详细解释一下哪个队列会发生什么？producer是否应该调用rd_kafka_consumer_poll来查看生成的消息？
浏览 0提问于2016-10-01得票数 0
2回答
我的卡夫卡消费者客户不断在频繁的民意调查之间进行再平衡，即使没有处理任何源记录。
此外，我在正确的位置上实现了consumer.pause()和consumer.resume()，以保持使用者从轮询到使用后端API隔离处理被调查的消费者记录的数量。不知道是什么引起了这个问题。
用户日志
2022-07-21 10:27:08; LOG_LEVEL="INFO"; SOURCE="MyKAFKAConsumer"; PLATFORM="Retail_Products"; SERVICE="MyKAFKAConsumer"; EVEN
浏览 41提问于2022-07-21得票数 0
我在Kubernetes上运行Kafka Connect (8-16个节点，自动标度)。我总共定义了44个连接器，每个Kafka主题一个(每个主题一个分区)。这些主题是由Debezium / Postgresql编写的。有3个Kafka节点。每个连接器的tasks.max设置为4。大部分连接器(但不是每个连接器！)有一个(总是一个)失败的任务，由于java.lang.IllegalStateException:没有当前分配的分区-0。
这里不是卡夫卡专家，请注意;)我假设有3个Kafka节点，因此3个工作人员做得很好，而第四个任务没有任何连接，所以失败了。但是为什么有时有4个任务运行得很好呢？
浏览 3提问于2019-03-24得票数 0
回答已采纳
我从日志中看到，665次使用了完全相同的消息。这一切为什么要发生？
我在日志里也看到了这个
Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member.
This means that the time between subsequent calls to poll() was longer than the configured session.timeout.ms, which typically implies
that
浏览 2提问于2018-08-08得票数 5
回答已采纳
我有一个主题，在Kafka 0.10.0.1集群中有10个分区。我有一个产生多个使用者线程的应用程序。就这个话题而言，我产生了5个线程。在我的应用程序日志中，我多次看到这个条目。
INFO :: AbstractCoordinator:600 - Marking the coordinator x.x.x.x:9092
(id:2147483646 rack: null) dead for group notifications-consumer
然后有几个条目说(Re-)joining group notifications-consumer.之后，我也看到一个警告说
Auto commit
浏览 1提问于2018-05-15得票数 8
回答已采纳
这里可能缺少一两个设置，但以KafkaListenerContainerFactory ackMode设置为手动设置，当我在接收时打印消息时，我会看到消息正在通过@KafkaListener方法流。我们不会在应用程序中的任何地方确认消息。感谢有人能指出问题的所在。提前谢谢。
使用spring版本-2.3.0- version
ListenerContainerFactory
KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<GenericData.Record, GenericData.Rec
浏览 6提问于2022-08-31得票数 1
如果我从poll()语句中的consumer中提取while True:，我就会发现poll()阻塞了。如果使用者对来自主题(offset = OFFSET_END)的消息是最新的，那么消费者如何执行它的阻塞poll()
消费者是否默认遵守酒吧/订阅的心态，在这种心态下，它会睡觉，等待来自经纪人的发布和广播/信号？
还是消费者在不停地自转，检查这个话题？
如果这有关系的话，我正在使用汇合python客户端。
谢谢!
浏览 4提问于2019-10-24得票数 1
回答已采纳
consumer.poll(Long.MAX_VALUE)；
这是为了允许处理(高延迟)完成。保持这样高的价值有什么副作用吗？
它是否阻止正在读取的分区中的其他消息？这会对再平衡产生不利影响吗？
浏览 2提问于2020-05-31得票数 1
当max.poll.interval.ms时间超过时，长处理记录会发生什么?它会在后台运行，并触发重新平衡。据我有限的理解，kafka使用者()服务被暂停/重新启动，并在再平衡期间将记录分配给组中的其他消费者。
浏览 4提问于2022-01-17得票数 0
我使用的是kafka 0.10.1.1，并混淆了以下3个属性。
heartbeat.interval.ms
session.timeout.ms
max.poll.interval.ms
heartbeat.interval.ms --这是在0.10.1中添加的，它将在民意测验之间发送心跳。session.timeout.ms -这是开始重新平衡，如果没有要求卡夫卡，并得到重置的每一项调查。max.poll.interval.ms --这是整个民意测验的结果。
但是，卡夫卡什么时候开始重新平衡？为什么我们需要这3？它们的默认值是什么？
谢谢
浏览 1提问于2017-05-16得票数 35
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
