如何让Kafka broker向消费者响应可恢复或不可恢复的错误？
为了让Kafka broker能够向消费者响应可恢复或不可恢复的错误，可以采取以下几种方法：
- 可恢复错误的处理：
- 重试机制：当Kafka broker发生可恢复的错误时，可以使用重试机制来尝试重新发送消息。这可以通过配置Kafka producer的重试次数和重试间隔来实现。
- 错误日志记录：Kafka broker可以将可恢复错误记录到日志中，以便管理员或开发人员进行故障排查和处理。
- 不可恢复错误的处理：
- 消费者错误处理：在消费者端，可以通过监控消费者组的健康状态来检测不可恢复的错误。如果发现某个消费者组出现不可恢复的错误，可以及时通知相关人员进行处理。
- 状态更新：Kafka broker可以在发生不可恢复错误时更新相关的状态信息，比如标记某个分区为不可用或进行故障转移。
需要注意的是，Kafka broker本身并没有提供特定的机制来处理可恢复或不可恢复的错误。这些处理方法主要依赖于应用程序的设计和开发。以下是一些与Kafka相关的腾讯云产品和服务：
- 云消息队列 CMQ：腾讯云提供的分布式消息队列服务，可用于异步通信、削峰填谷等场景。
- 产品介绍链接：云消息队列 CMQ
- 云数据库 CynosDB：腾讯云提供的高性能、高可用的数据库服务，可用于存储和管理Kafka相关的数据。
- 产品介绍链接：云数据库 CynosDB
- 云原生容器服务 TKE：腾讯云提供的容器化部署和管理服务，可用于将Kafka部署在容器中，并进行弹性伸缩和自动化管理。
- 产品介绍链接：云原生容器服务 TKE
以上产品和服务都是腾讯云提供的解决方案，可以帮助用户在云计算领域构建可靠、高效的系统。
相关·内容
我们使用的是kafka-streams 2.3.1，我刚刚注意到，如果broker宕机，streams应用程序似乎满足于尝试永远保持连接。 new KafkaStreams(createTopology(), properties()).start() o.apache.kafka.clients.NetworkClient - [AdminClient clientId=test] Connection to node -1 (broker/127.0.0.1:9092) could not be established. Broker may not be available. 在这种
浏览 41提问于2019-12-19得票数 2
我有两台机器localhost和192.168.1.110来运行两台独立的单机kafka。
(1)在localhost上，我运行：
主控台kafka_2.11-0.10.0.0bin/kafka-➜-Producer.sh --broker-list 192.168.1.110:9092 --topic test这是一条消息2016-08-24 18:15:27,441向topic test发送消息时出错: null，value: 2字节错误:包含1条记录的(org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) or
浏览 13提问于2016-08-24得票数 0
我写了一个node-kafka consumer。在极少数情况下，我使用一个组id启动kafka客户端，当某些偏移量可用时使用，但不再可用-导致"offsetOutOfRange“事件被调用。在这种情况下，推荐的行为是什么？记录错误并退出？有没有办法恢复？我总是想从上次提交的偏移量开始运行zookeeper (如果存在且可用)。
client = new kafka.Client(ZOOKEEPER_URLS),
consumer = new Consumer(client, [], {
groupId: GROUP_ID,
fromOffset: true
浏览 10提问于2016-08-01得票数 1
回答已采纳
这是关于卡夫卡MirrorMaker工具。
我在两台机器上配置了卡夫卡。
来源：
目的地: vm仅位于源端
卡夫卡在来源和目的地都是同样版本的卡夫卡卡夫卡_2.11-0.9.0.0
在源和目的地，各自的动物园管理员和卡夫卡服务器正在运行。
使用MirrorMaker工具，我想从源到目的地复制/制作主题镜像。
下面是我使用的命令：
./bin/kafka-run-class.sh kafka.tools.MirrorMaker --consumer.config ./config/mirror_consumer.properties --producer.config .
浏览 7提问于2016-05-06得票数 2
我对卡夫卡很陌生。我有以下架构：
1) 2 servers runing application logics and database, can I write kafka producer on these servers wrapped with docker container ?
2) 1 server reserved for kafka broker and zookeeper
3) 1 sever reserved for kafka comsumer
我很困惑
1) whether I can run kafka producer, broker and consumer
浏览 6提问于2015-11-02得票数 0
回答已采纳
有一台公网ip已知的云服务器，kafka broker在其中启动并运行。如何从我的笔记本电脑生成和消费发送给此kafka broker的消息。
浏览 1提问于2017-09-13得票数 0
1回答
这里有一个奇怪的问题，卡夫卡在一个带有一个动物园管理员实例的2个代理设置中，都在AWS上。每个代理位于单独的可用性区域(AZ)。孤独的动物园管理员实例与其中一个经纪人共享AZ。我试图模拟一个滚动重新启动的情况，我们希望数据流到Kafka，即使在经纪人倒闭的时候。
当另一个AZ经纪人被拿下的时候，我能看到一个平稳的过渡到与Zookeeper共享AZ的经纪人。但是，当我在与Zoo门将共享的AZ上取下Kafka broker时，其他AZ代理将不会使用数据。不过，我认为领导人的分配会按预期的方式进行。我不确定这是否是我们的设置，卡夫卡，或动物园管理员的问题。
我们为代理间通信启用了SSL，而且生产者也
浏览 1提问于2016-06-15得票数 1
以Kafka生产者为例：
const { Kafka, logLevel } = require('kafkajs');
const kafka = new Kafka({
clientId: 'my-app',
brokers: ['kafka1:9092', 'kafka2:9092'],
logLevel: logLevel.ERROR,
});
const run = async () => {
const producer = kafka.producer();
await produce
浏览 3提问于2020-10-09得票数 3
我正试图在两个物理机器上安装一个HLF (v2)网络。机器1：
1订购者2对等点2 couchdb 2动物园管理员2 kafka
机器2：
1秩序者2对等点2 couchdb 1动物园管理员2 kafka
集装箱在这两台机器上运行。
但是，当我试图在机器1上创建通道时
peer channel create -o orderer.example.com:7050 -c $CHANNEL_NAME -f ./channel-artifacts/channel.tx --tls --cafile /opt/gopath/src/github.com/hyperledger/fabric/peer/
浏览 4提问于2020-06-09得票数 0
1回答
我在一个Kafka集群中消费一个Kafka主题(保留21天)，作为消费者(最早/从开始)连续15天与x消费者组在一起，在第15天，生产者团队停止生产，我在我这边停止消费者，以确保没有剩余的消息可供消费。然后Kafka集群也被关闭了。然后，在第16天，Kafka Cluster启动了，Producer在第23天启动了他的producer，我也启动了我自己的消费者。但当我开始的时候，我从一开始就收到了信息，而不是从我遗漏的地方，尽管我是在同一个x消费者群体中消费。所以我的问题是为什么会发生这种情况？Kafka Broker是否丢失了消费群体信息？
浏览 29提问于2020-07-24得票数 0
3回答
kafka-console-consumer.bat --bootstrap-server localhost:2181 --topic baeldung kafka-console-producer.bat --broker-list localhost:9092 --topic baeldung
消息不会从生产者传递到消费者。
浏览 0提问于2018-12-20得票数 1
3回答
我有一个与Kafka通信的spring引导应用程序。
我通过注入环境变量在生产中配置此应用程序。
对于kafka，我可以使用环境变量配置大多数东西--引导服务器、ssl信任存储位置、ssl信任存储密码、组id、主题，例如：
SPRING_KAFKA_SSL_TRUSTSTORE-LOCATION: "file:/opt/app/jks/totally_real_file.jks"
SPRING_KAFKA_SSL_TRUSTSTORE-PASSWORD: "hunter2"
因为我可以使用环境变量配置ssl密钥存储库和密码，所以我假设可以配置SSL协议和安全协议
浏览 2提问于2018-07-13得票数 3
我正在尝试让我的一个服务在码头网络中在本地运行，但是一直让“消费者启动失败:连接被拒绝-连接(2)对于127.0.0.1:9092”的异常。
zookeeper:
image: wurstmeister/zookeeper:3.4.6
restart: always
expose:
- "2181"
kafka:
depends_on:
- zookeeper
environment:
KAFKA_ADVERTISED_LISTENERS: INSIDE://kafka:9093,OU
浏览 15提问于2020-05-24得票数 1
在上查看此文档
Zookeper在端口2181 Broker/Kafka启动，时间为9092。
启动生成器发送消息
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Hello-Kafka // producer is producing directly to broker not zookeeper which is correct
启动使用者接收消息
bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic Hello-
浏览 15提问于2017-07-11得票数 3
回答已采纳
1回答
我有以下集群：
Kafka ->一些日志收集器-> Elasticsearch
我的问题是如何选择最有效的日志收集器(或者其他一些允许管理Kafka和ElasticSearch之间数据流的软件)。
我正在尝试从Logstash、Fluentd和Confluent的Kafka Elasticsearch连接器中进行选择。我面临的主要问题是无法在写入Elasticsearch端点时在Kafka中回滚偏移。
例如，logstash说“如果启用400和404错误将发送到死信队列( DLQ )。如果未启用DLQ，则将发出日志消息，并删除事件”()。如果我有这样的错误，logstash将继续读
浏览 1提问于2018-10-19得票数 0
在kafka borker中，保留期为零是否有意义？ 我们希望通过kafka broker将消息从生产者快速转发到消费者。从broker机器上的buffercache/pagecache，而无需刷新到磁盘。我们不需要复制，并假设我们的代理永远不会崩溃。
浏览 17提问于2019-03-16得票数 4
回答已采纳
在Apache中的分区主题上，我有点困惑。因此，我正在绘制一个简单的用例，我想知道在不同的场景中会发生什么。所以这里是：
我有一个主题T，它有4个分区( TP1、TP2、TP4和TP4 )。
假设我有8条消息( M1 to M8 )。现在，当我的制作人将这些消息发送到主题T时，Kafka broker将如何在以下情况下接收这些消息：
场景1:只有一个kafka broker实例具有前面提到的分区的主题T。
场景2:有两个kafka broker实例，每个节点具有相同的主题T和前面提到的分区。
现在假设kafka broker实例1下降了，消费者将如何反应？我假设我的消费者是从broker实例1
浏览 5提问于2016-01-13得票数 0
回答已采纳
我正在尝试在不同的机器上设置多节点Kafka-0.8.2.2集群，包括一个生产者、一个消费者和三个代理。
在producer上创建主题时，我得到的错误是org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /brokers/ids。完整的控制台输出可用。Kafka Producer的日志中没有错误。
我用来运行Kafka的命令是：
./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-facto
浏览 7提问于2016-03-02得票数 1
回答已采纳
1回答
我对Golang和Kafa并不熟悉，所以这似乎是个愚蠢的问题。
当我的卡夫卡消费者第一次连接到卡夫卡服务器之后，为什么连接到Kafka服务器和接收到第一条消息之间会有延迟(~ 20秒)？
它在consumer.Messages()之前打印一条消息，并为收到的每条消息打印另一条消息。~20秒延迟在第一fmt.Println和第二fmt.Println之间。
package main
import (
"fmt"
"github.com/Shopify/sarama"
cluster "github.com/bsm/sarama-
浏览 0提问于2018-12-07得票数 0
我执行shell脚本kafka-topics.sh将分区数从3改为10。
./bin/kafka-topics.sh --zookeeper xxx.xx.xxx.xx:2181/my-broker-0 --alter --topic target-topic-example-1 --partitions 10
我一执行这个命令，Kafka集群就出现了错误。
[2018-11-20 14:45:29,954] INFO Created log for partition [target-topic-example-1,5] in /data/kafka/brokers with propert
浏览 0提问于2018-11-20得票数 0
我试图运行简单的卡夫卡生产者消费者的例子，在HDP，但面临以下例外。
[2016-03-03 18:26:38,683] WARN Fetching topic metadata with correlation id 0 for topics [Set(page_visits)] from broker [BrokerEndPoint(0,sandbox.hortonworks.com,9092)] failed (kafka.client.ClientUtils$)
java.nio.channels.ClosedChannelException
at kafka.netw
浏览 0提问于2016-03-03得票数 0
回答已采纳
1回答
我已经创建了三个单独的容器:一个为Kafka，第二个为Porducer(Streamer)，最后一个是为消费者使用的码头-合成。
version: "3"
services:
zookeeper:
image: confluentinc/cp-zookeeper:latest
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_TICK_TIME: 2000
networks:
- stream-network
kafka:
image: co
浏览 0提问于2020-06-01得票数 0
2回答
嗨，我是卡夫卡CLI的工作，以了解卡夫卡如何工作。我对消费者群体感到困惑。我已经创建了三个分区的主题。我将创建生产者来为主题提供一些数据。我第一次添加一些数据如下所示。
kafka-console-producer --broker-list 127.0.0.1:9092 --topic users
>user1
kafka-console-producer --broker-list 127.0.0.1:9092 --topic users
>user2
kafka-console-producer --broker-list 127.0.0.1:9092 --topic u
浏览 7提问于2019-11-15得票数 2
回答已采纳
我正试图在ksqldb之上构建一个应用程序。
假设我会有一个简单的制作人：
package main
import (
"fmt"
"github.com/rmoff/ksqldb-go"
"net/http"
)
var client = ksqldb.NewClient("http://localhost:8088", "", "").Debug()
func init() {
offset := `SET 'auto.offset.reset
浏览 14提问于2022-02-14得票数 0
回答已采纳
我正在使用下面的命令尝试使用kafka主题中的消息。但这是行不通的。
bin/kafka-console-consumer.sh --bootstrap-server BROKER_URL --topic TOPIC_NAME --from-beginning
我的同事使用了下面的查询，它正在工作。
bin/kafka-console-consumer.sh --bootstrap-server BROKER_URL --topic TOPIC_NAME --offset 0 --partition 0
我不知道这两个命令之间有什么区别，为什么第二个命令工作，而不是第一个命令。
浏览 0提问于2019-01-06得票数 1
1回答
我想知道有没有Kafka Spout支持安全的Kafka Broker。来自apache storm的KafkaSpout不支持SSL。
public KafkaSpout(SpoutConfig spoutConf) {
_spoutConfig = spoutConf;
}
下面提到的Kafka没有接受SSL Kafka生产者/消费者支持的任何参数。
public KafkaConfig(BrokerHosts hosts, String topic) {
this(hosts, topic, kafka.api.OffsetRequest.DefaultClien
浏览 3提问于2017-03-30得票数 1
当file节拍输出数据到kafka时，文件日志中有许多警告消息。
。。
*WARN producer/broker/0 maximum request accumulated, waiting for space
*WARN producer/broker/0 maximum request accumulated, waiting for space
。。
在我的文件节拍配置中没有什么特别之处：
。。
output.kafka:
hosts: ["localhost:9092"]
topic: "log-oneday"
。。
我还在kafka中更
浏览 2提问于2017-01-23得票数 1
1回答
动物园管理员和卡夫卡正在用
船坞组成-d
---
version: '2'
services:
zookeeper:
image: confluentinc/cp-zookeeper:6.1.0
hostname: zookeeper
container_name: zookeeper
ports:
- "2181:2181"
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_TICK_TIME: 2000
broker:
浏览 5提问于2021-11-24得票数 0
回答已采纳
1回答
Kafka未连接到下面的问题
、、、、
2021-09-09 21:42:40.039 WARN 14636 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient : [Consumer clientId=consumer-mygroup-1, groupId=mygroup] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2021-09-09 21:42:40.039 WARN 14636 ---
浏览 7提问于2021-09-09得票数 1
我正在尝试用kafka实现一个多消费者-单生产者的设置。我创建了一个包含3个分区的主题，如下所示：
./kafka-create-topic.sh --topic stream.main.out --zookeeper
localhost:2181 --partition 3
producer的设置如下：
props = new Properties();
props.put("serializer.class", "kafka.serializer.StringEncoder");
props.put("broker.l
浏览 0提问于2018-03-23得票数 0
1回答
我们的问题是内部属性kafka.bootstrap.servers不会被kafka:9092覆盖。我们的容器仍然试图访问localhost:9092。
我们在docker-compose.yml文件中将每个微服务的环境设置为: kafka.bootstrap.servers=kafka:9092，并且每个MS的属性也是相同的。
有什么方法可以覆盖这个属性吗？
编辑:我们的docker-compose.yml文件的重要部分：
services:
zookeeper:
image: strimzi/kafka:0.19.0-kafka-2.5.0
command:
浏览 11提问于2021-06-02得票数 0
回答已采纳
是否有任何选项或配置可以方便单个Kafka消费者同时消费来自两个不同集群的消息？在创建生产者和消费者时，我提到这两个集群都是逗号分隔的。我一直在观察消费者只使用来自单个集群的消息。
请参见下面的说明:消费者集群配置为侦听集群: C1 -1: Zookeeper-1 with Broker-1 Cluster-2: Zookeeper-2 with Broker-2
我正在寻找一种解决方案，其中消费者C1可以同时使用来自集群1和集群2的消息。
浏览 0提问于2019-03-18得票数 0
下面是场景:我希望IIS能够从我的代码提供的自定义源中为URL提供服务。其中一个例子是从SQL数据库而不是文件系统提供文件(不过，我知道关于SQL服务器文件表的情况，这些不是合适的解决方案)。
我知道handlers (*.ashx)，但不知怎么的，这些处理程序从来不使用任何东西。我从来没有让IE的简历下载功能与他们一起工作，更不用说移动设备上的流媒体视频文件了。IIS中必须有一个t代码，它决定了如何处理所有这些情况，我不想从头开始重写这些代码，并冒着丢失一些关键内容的风险。
我正在寻找的(我认为这很简单)是一些回调，它要求数据的某些部分，让IIS处理HTTP响应细节。如果这是不可能的，那么还
浏览 3提问于2017-05-13得票数 0
我试着让两个消费者订阅一个EmbeddedKafkaBroker。第一个成功了，第二个失败了。@EmbeddedKafka和@ClassRule brokers都失败了。
@RunWith(SpringRunner.class)
@SpringBootTest
@EmbeddedKafka(topics = { "topic" })
public class AnnotationEmbeddedKafkaTest {
@Autowired
private EmbeddedKafkaBroker broker;
@Test
public voi
浏览 17提问于2021-05-18得票数 0
我们有卡夫卡集群，集群包括7卡夫卡经纪人，
版本详细信息：
Kafka machines are installed on rhel version 7.2
Kafka version is 0.1x
Zookeeper version: 3.4.10
schema-registry version: 4.0.0
每台Kafka代理机器包括以下服务
Kafka broker
Zookeeper server
Schema registry
现在我们要从当前的Kafka集群中删除2 Kafka代理机器(我们要删除的卡夫卡机器是- 06/07 ( Kafka版本- 0.1X，)
因此，集群最终应
浏览 2提问于2020-08-19得票数 1
回答已采纳
1回答
Kafka集群提供了高可用性，但它是否也提供了一些容灾保护？ 具体地说，假设您的一个主题文件在一台服务器上以某种方式损坏或删除，您是否可以通过集群中其他服务器上的主题文件来恢复？
浏览 25提问于2021-10-29得票数 0
我对库伯奈特斯很陌生，对卡夫卡也很陌生。
我的目标是让我的Kafka实例正确地连接到我的代理，以便它能够接收消息。
到目前为止，我已经创建了一个部署，并将其封装在一个服务中。(据我所知，这是卡夫卡经纪人能够把动物园管理员简单地称为“动物园管理员:2181”的必要条件。)
我曾希望，建立卡夫卡连接实例将是同样简单的。我将代理封装在一个服务中，并通过名称将Connect实例指向它。
worker.properties
bootstrap.servers=kafka-broker:9092
但是，Connect除此例外情况外会崩溃：
java.io.IOException: Can'
浏览 0提问于2019-07-03得票数 1
回答已采纳
1回答
将Instabug与我的swift 2.3应用程序集成在一起，但是看到了Instabug accepting NSException的错误报告api如下，
Instabug.reportException(e)
但是在Wave2.3中，我在catch块中获得了NSError实例。
do {
}
catch let error as NSError {
}
我想把我在catch块中遇到的错误报告给Instabug。请让我知道我怎么做的。
浏览 5提问于2016-10-27得票数 0
我们有一个HDP集群，有3个kafka经纪人(来自hortonworks )。
我们希望运行kafka控制台使用者，以便从带有特定偏移量的主题中获得一条消息。
/usr/hdp/current/kafka-broker/bin/kafka-console-consumer.sh --zookeeper zoo01:2181 --topic lopet.lo.pm--partition 0 --offset 34537263 --max-messages 1
但我们得到了以下信息：
我们哪里错了？
Using the ConsoleConsumer with old consumer is d
浏览 3提问于2019-12-11得票数 5
回答已采纳
我知道什么是生产者和消费者。但官方文件上说
它是流媒体平台。
它是企业消息传递系统。
Kafka还拥有从数据库和其他系统导入和导出数据的连接器。
什么意思？
我知道生产商是将数据发送给Kafka Broker的客户端应用程序，消费者也是从Kafka Broker读取数据的客户端应用程序。
但我的问题是，消费者能否将数据推入卡夫卡经纪商？
据我所知，我认为如果消费者想把数据推入卡夫卡经纪人，它就会成为一个生产者。对吗？
浏览 6提问于2017-06-02得票数 1
回答已采纳
如果我有多个经纪人，我的制片人应该使用哪个经纪人？是否需要手动切换代理以平衡负载？此外，为什么使用者只需要一个动物园管理员端点而不是代理端点？
教程中的快速示例：
> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
> bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
浏览 4提问于2014-03-16得票数 36
回答已采纳
1回答
在我使用golang客户端提交client之后。
./kafka-consumer-offset-checker.sh --zookeeper=localhost:2181 --topic=my-replicated-topic --group=ib --broker-info
Group Topic Pid Offset logSize Lag Owner
ib my-replicated-topic 0 1
浏览 4提问于2015-05-20得票数 0
我正在学习kafka和vertx，我理解了以下的说法，
1.Kafka模块允许接收其他Vert.x眩晕事件，并将这些事件发送给Kafka broker。
2.应用程序使用Vertx总线向Kafka模块发送消息。
3.Kafka模块作为制片
任何人让我知道他们是如何编程的，将是非常有帮助的。谢谢。
我在这里找到了源代码，但我正在寻找一个更简单的例子。
浏览 3提问于2014-10-05得票数 0
回答已采纳
我对Python相当陌生，刚刚开始使用Kafka，所以如果我错了，请原谅我的术语。
所以我有一个基于Django的web应用程序，在这个应用程序中，我通过Kafka生产者在同一个过程中发送一个json消息。然而，在以务实的方式创建主题的同时，我也在为该特定主题的单独过程中开始(订阅)一个新的使用者。
#Consumer code snippet
if topic_name is not None :
#Create topic
create_kafka_topic_instance(topic_name)
#Initialize a cons
浏览 0提问于2019-04-08得票数 0
我使用在kubernetes部署了kafka。但是，在使用kafka消费者消费时，我得到了以下错误：
SEVERE: Failed to resolve default logging config file: config/java.util.logging.properties
[10:23:00] __________ ________________
[10:23:00] / _/ ___/ |/ / _/_ __/ __/
[10:23:00] _/ // (7 7 // / / / / _/
[10:23:00] /___/\___/_/|_/
浏览 1提问于2018-09-25得票数 0
回答已采纳
我正在考虑使用Kafka连接器和创建我自己的Kafka消费者/生产者来将一些数据从Kafka转移到Kafka，我看到了Kafka连接器在可扩展性和容错方面所提供的价值。但是，如果“任务”由于某种原因而失败，我无法找到连接器的确切行为。以下是几种场景：
对于接收器连接器(S3-Sink)，如果它(任务)成功地将数据发送到目标(例如由于网络问题)失败(在所有重试之后)，那么工作人员会发生什么情况？它会坠毁吗？它能够再消费同样的数据从Kafak以后吗？
对于源连接器(JDBC )，如果它不能发送给Kafka，它以后会重新处理相同的数据吗？这取决于来源是什么？
以上问题的答案是否取决于
浏览 0提问于2018-08-22得票数 1
2回答
Kafka Broker有3个分区。
卡夫卡消费者实例的计数是3。
突然，一个消费者的例子死了。
我知道，如果一个Kafka消费者实例死亡，Kafka代理正在进行再平衡，而另一个消费者实例将被分配到该分区。
我想知道这样的假设是否正确:另一个实例使用它最初使用的所有分区，然后分配和使用死分区。
(我是否必须在客户端代码中实现ConsumerRebalanceListener？)
如果是这样的话，在使用消息时会有什么延迟吗？
谢谢。
浏览 3提问于2017-05-17得票数 1
我正在构建一个websocket后端，它将连接到一个主题(只有一个分区)，并从最早的位置消费数据，并不断消费新数据，直到websocket连接断开。一次可以存在多个websocket连接。 为了确保从头开始的所有数据都被使用，每次建立websocket连接时，我都会创建一个新的消费者组并订阅该主题 const Kafka = require('node-rdkafka')
const { v4: uuidv4 } = require('uuid')
const kafkaConfig = (uuid) => ({
'group.id'
浏览 35提问于2021-11-13得票数 0
回答已采纳
针对消费者 bin/windows/kafka-console-consumer.bat --bootstrap-server localhost:9092 --top
irer 对于制片人 bin/windows/kafka-console-producer.bat --broker-list localhost:9092 --topic test 因为我认为对于生产者，我们也只能选择一个代理，从这个代理服务器中，我们可以找到所有的分区信息和领导者，所以我不认为broker-list是真正需要的。
浏览 192提问于2020-04-30得票数 1
回答已采纳
我试图用kafka处理Meetup流api。
这一进程如下：
启动动物园管理员
启动卡夫卡
启动卡桑德拉
运行rsvp_producer.py
运行rsvp_consumer.py <-我在这里。
当我尝试运行rsvp_consumer.py时，我在Mac上的终端中得到了下面的错误。
$ python rsvp_consumer.py
Traceback (most recent call last):
File "rsvp_consumer.py", line 12, in <module>
metadata_bro
浏览 4提问于2016-02-28得票数 2
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
