在Kafka中有没有办法检测到被删除的消费群？
在Kafka中，可以通过使用Kafka的内置工具和功能来检测被删除的消费群。
- 使用Kafka自带的工具：Kafka提供了一些命令行工具，如kafka-consumer-groups.sh，可以用于管理和监控消费群。通过运行该工具的describe命令，可以列出所有消费群的详细信息，包括消费群的消费者列表。如果一个消费群被删除，它将不再显示在列表中。
- 使用Kafka的API：Kafka提供了一些API，可以通过编程方式检测被删除的消费群。例如，可以使用Kafka的AdminClient API来获取消费群的列表，并检查某个消费群是否存在。如果一个消费群被删除，它将不再在列表中出现。
需要注意的是，Kafka本身并没有提供直接的机制来检测被删除的消费群。上述方法只能通过间接的方式来检测消费群是否存在。另外，Kafka的消费群管理是分布式的，消费群的删除可能需要一定的时间才能在整个集群中生效。
推荐的腾讯云相关产品：腾讯云消息队列 CMQ（Cloud Message Queue）是一种高可靠、高可用的分布式消息队列服务，适用于大规模分布式系统的消息通信。CMQ提供了消息队列、主题订阅、消息发布等功能，可以满足各种异步通信和解耦需求。
腾讯云产品介绍链接地址：https://cloud.tencent.com/product/cmq
相关·内容
显然，在早期的kafka版本中，某个消费群体的消费者列表存储在zookeeper中。对于最新的kafka版本，这些信息存储在哪里？
浏览 18提问于2019-12-21得票数 2
回答已采纳
我们有一个使用kafka作为输出的spark批处理应用程序。有没有办法监控Kafka在使用spark时的消费群体滞后？我们查看了spark文档，发现spark总是创建一个随机的groupId，并使用检查点目录来控制偏移量。你们有没有推荐一种方法或应用程序来衡量这个过程的滞后？我们找不到做这件事的方法。
浏览 16提问于2019-11-29得票数 1
回答已采纳
有一个Kafka服务器使用Kafka来存储消费者组的偏移量，我想知道__consumer-offsets topic是否会有一个针对已删除的消费者组的事件？或者我如何订阅此活动而不询问Kafka是否存在特定的消费群体
浏览 42提问于2020-09-12得票数 0
我是新来的。我遇到了这个问题，我的应用程序每5分钟运行一次，从云中有一些记录的blob中读取数据，并将这些记录处理到kafka主题中。问题是，现在它在单个实例中运行，没有问题，但如果我计划在多个实例中运行，如果两个实例同时运行相同的记录，那么在kafka主题中将会有重复的记录，因为代码库是相同的。在spring batch中有没有什么特性可以帮助你克服这个问题，或者其他不会产生重复的方法。
提前感谢Vamsi
浏览 7提问于2020-05-06得票数 0
所有的人，请原谅我，我是一个新手刚刚开始的Kafka。目前，我正在阅读Kafka关于的文档。
如文件所述。
对于传统的信息系统。它们不能缩放消息处理。因为
发布-订阅允许将数据广播到多个进程，但由于每条消息都发送到每个订阅服务器，因此无法扩展处理。
我觉得这对我来说很有意义。
但是对于Kafka来说。文档说，即使在发布-订阅模式下，Kafka也可以扩展消息处理。(如果我错了，请纠正我。谢谢。)
卡夫卡的消费群体概念概括了这两个概念。与队列一样，使用者组允许您将处理划分为进程的集合(使用者组的成员)。与发布订阅一样，Kafka允许您向多个消费群体广播消息。
卡夫卡模型的优点在于，每个主
浏览 1提问于2017-07-16得票数 0
1回答
在我们的kafka集群中(基于HDP版本- 2.6.5，kafka版本为1.0 )，我们希望删除以下消费者组
/usr/hdp/2.6.5.0-292/kafka/bin/kafka-consumer-groups.sh --list --bootstrap-server kafka01:6667| grep -i test
Note: This will not show information about old Zookeeper-based consumers.
test_group_0e1e37b0-603e-4ffd-9b21-90a3adf597ca
但当我们试图删除它时，我
浏览 13提问于2022-02-21得票数 1
回答已采纳
例如，我的kafka消费者(在consumer Group 1中)正在阅读来自Kafka主题A的消息。
现在，如果该消费者在失败之前消费了12条消息。
当消费者重新启动，并且现在它有不同的消费者组(即消费者组2)时，问题1 -？在重启时，它会从偏移量(或位置)中停止的地方继续，因为该偏移量是由Kafka和/或ZooKeeper存储的，还是会从第一条消息开始消费消息。
问题2->有没有办法确保在重启时(当消费者有不同的消费群体时)，它仍然从重启之前停止的地方开始消费？
为了给你提供上下文，我正在尝试在收到关于kafka主题的消息时更新每个节点/服务器中的内存缓存。为此，我对每个节点/服务器
浏览 1提问于2021-02-04得票数 0
1回答
我的问题是如何使用Spring Cloud Stream Kafka管理多实例。 让我解释一下，在Spring Cloud Stream微服务上下文(eureka，configserver，kafka)中，我希望拥有相同微服务的两个实例。当我在GIT Repository中更改配置时，configserver (通过webhook)会将一条消息推送到Kafka主题中。 如果我在我的微服务中使用相同的group-id，那么两个实例中只有一个会收到通知，并重新加载他的spring上下文。但我需要刷新所有实例... 因此，为了做到这一点，我配置了一个唯一的组id：${spring.applicat
浏览 44提问于2020-07-17得票数 2
回答已采纳
1回答
我在卡夫卡创建了大量的消费群体。当我检查状态时，它们中的大多数都是空状态，并且没有活动的使用者。
我有两个问题
自动删除消费者组在Kafka中是支持的，我们可以配置一个时间段来删除空的消费组吗？是否有支持自动删除空消费者组的特定版本？
浏览 2提问于2020-12-07得票数 4
回答已采纳
在Kafka中，每个消费者组由许多消费者实例组成，以实现可伸缩性和容错。 kafka消费者可以属于多个消费群体吗？
浏览 30提问于2020-08-19得票数 1
回答已采纳
我让Kafka Connect JDBC接收器连接器写入不同的数据库，我想限制到一个数据库的流量。Kafka quotas功能可以为客户端ID设置consumer_byte_rate配额，但Kafka Connect客户端ID看起来像consumer-1234，并动态分配给连接器。因此，如果我的接收器连接器被重新平衡，它将被分配所有新的客户端ID。我尝试使用我的接收器连接器使用者组ID作为客户端ID来设置配额，但不起作用。有没有办法给Kafka Connect消费群设置配额？
浏览 16提问于2019-12-18得票数 0
1回答
我正在创建一个Perl无状态水平可伸缩的微服务，并正在寻找一个高级Kafka消费者组实现。
我找到了这个库，它有一些低级的功能，但是有很多限制。
必须显式地声明分区。
的返回值
$the>fetch_offsets( $topic，# topic 0，$group )；是一个散列，不能在以下文件中使用：
$offsets>commit_偏移( $topic，0，$offsets1，$group )；方法。
等等..。
在Kafka的新版本中，消费者组完全由Kafka来处理，而不需要一个动物园管理员客户端，以及Java、Python、Scala中的客户端.只需状态
浏览 3提问于2019-09-22得票数 2
1回答
我在一个Kafka集群中消费一个Kafka主题(保留21天)，作为消费者(最早/从开始)连续15天与x消费者组在一起，在第15天，生产者团队停止生产，我在我这边停止消费者，以确保没有剩余的消息可供消费。然后Kafka集群也被关闭了。然后，在第16天，Kafka Cluster启动了，Producer在第23天启动了他的producer，我也启动了我自己的消费者。但当我开始的时候，我从一开始就收到了信息，而不是从我遗漏的地方，尽管我是在同一个x消费者群体中消费。所以我的问题是为什么会发生这种情况？Kafka Broker是否丢失了消费群体信息？
浏览 29提问于2020-07-24得票数 0
1回答
如果我们有一个大文件，我想知道这将如何影响未检测到错误的概率，特别是在CRC中。我知道未检测到的错误率(分组或块) is=比特*误码率* 0.5^k，其中K是CRC的FSC。在CRC32k是31， 从这个等式和下面的图片可以看出，对于不同的数据包大小，数据包大小不会影响未检测到错误的概率。假设我们有1,000,000个包，每个包都有2^(-32)未检测到错误的概率，我如何计算整个1Petabyte文件的未检测到错误的概率？ ?
浏览 54提问于2020-07-13得票数 0
1回答
我使用的是单节点Kafka broker (v0.10.2)硬件(8核，16 GB内存，1 TD HardDisk)和动物园管理员(v3.4.8)。我有一个有200个分区的主题，其中包含300万条消息。它花了5天的时间来完全处理所有的消息，一旦消息被处理，即Kafka-消费群。of在主题的所有分区中显示出0滞后，6小时后我停止了使用者.but，它显示了200万条消息的滞后，我发现它是重复的消息。这件事经常发生。我的补偿储存在卡夫卡经纪人身上。我的服务器配置是：
broker.id=1
delete.topic.enable=true
#listeners=PLAINTEXT://:9092
#
浏览 0提问于2017-05-19得票数 1
回答已采纳
1回答
我正在开发一个使用Akka Kafka连接器的Kafka消费者应用程序。我希望消费者并行处理消息。我应该选择哪个消费群体？？如何在消费者端配置并行度？
浏览 16提问于2020-05-06得票数 0
在合流的kafka rest代理中，我们可以得到特定消费群体的最后提交偏移量，但是如何才能得到主题的最新偏移量来计算滞后值。
浏览 2提问于2019-07-11得票数 0
我正在尝试通过终端中的以下命令来描述kafka消费群： ./kafka-consumer-groups.sh --bootstrap-server $IP:$PORT --describe --group $GROUPID 我可以通过以下命令获取$GROUPID： ./kafka-consumer-groups.sh --bootstrap-server $IP:$PORT --list 回应是： b'TestProcess' 但在尝试通过以下命令获取有关该组的详细信息后： ./kafka-consumer-groups.sh --bootstrap-server $IP:$
浏览 142提问于2019-05-26得票数 2
回答已采纳
我正在运行多个kafka使用者，每个用户都在自己的服务器上，从远程kafka集群上定义的相同主题中读取数据。50台服务器中只有5台成功运行使用者进程。其余的日志都没有启动--以下是日志：
我，2017-06-20T20:25:13.524287 #13302 INFO --登记为web1:d671ebce- as 76-4917-b504- for 5924994f2 I，2017-06-20T20:25:13.529857 #13302 INFO -6例已注册: web11:86472d4d-cc95-4a5d-bc52-d1c97a8a9c34，web1:d671ce-bf76-4917
浏览 3提问于2017-06-20得票数 0
回答已采纳
1回答
我正在做一些关于Kafka和Avro的研究，我有一个关于如何使用Kafka处理异构消息的问题。
假设我们有两个Avro模式：EmailSent和UserLoggedIn。卡夫卡的最佳消费方式是什么？
为每个架构创建一个主题？但如果我们有更多的模式听起来不是个好主意..。
将模式名称(或像email_sent或user_logged_in这样的标识符)存储在KeyedMessage键中，让每个使用者按键进行筛选？但是每个消费者都要消耗掉所有的事件，这可能是一种开销.
(另一种:)
我希望我的疑虑是清楚的。提前感谢您的帮助！
干杯,
镭。
浏览 2提问于2015-07-08得票数 0
我们使用的是之前版本的Kafka(0.8.X)，其中所有偏移量细节都存储在ZooKeeper中。现在我们转向了卡夫卡(0.10.X)的新版本，所有的主题偏移细节都存储在Kafka本身中。我们必须将所有的主题偏移细节转移到ZooKeeper，以获取生产中现有的应用程序。卡夫卡安装在Windows机器上。我们不能从窗口运行kafka-消费群。请建议如何将提交的偏移量从ZooKeeper迁移到Kafka。
浏览 0提问于2018-02-08得票数 0
1回答
卡夫卡·努布来了！我有两个问题：
1)将少数卡夫卡消费者作为同一个卡夫卡消费群体的一部分，在不同的库伯奈特豆荚上运行。这些吊舱所消耗的主题是压缩主题。现在，让我们假设其中一个豆荚下降，并上来一段时间。现在我的问题是，消费者会不会收到卡夫卡这个主题的所有信息？或者，它是否只收到在它从失败中回来后到达的话题？
2)我了解Kafka消费者从从“提交的偏移”开始的分区中接收主题。在“压缩主题”的情况下，它将如何工作，因为Kafka只发送带有最新偏移量的事件。
浏览 0提问于2018-04-05得票数 1
回答已采纳
1回答
我有一个用例，需要向水平可伸缩、无状态的应用程序集群中的所有节点广播消息，我正在考虑使用Kafka。因为集群的每个节点都需要接收主题中的所有消息，所以集群的每个节点都需要有自己的使用者组。
这里可以假设消息的数量并不高，每个节点都不能处理所有的消息。
为了用Kafka实现这一点，我将在从主题消费时使用使用者进程的instanceId (或某些唯一标识符)作为使用者组id。这将推高消费群体的数量。随着重新部署的完成，新的使用者组将开始。
，在任何给定的时间，我最多能拥有多少活跃的消费者群体？在其他瓶颈(如带宽等)出现之前，消费群体的数量会成为瓶颈吗？，，在频繁部署使用者应用程序时，会出现活跃的使
浏览 5提问于2021-07-15得票数 0
例如。这是一个偏移量为100的消费者组，而groupId是' group -0‘。
我用卡夫卡-消费者团体来删除那个消费群体。
kafka-consumer-groups --bootstrap-server <address> --delete --group group-0 --execute
它还取得了成功：
Deletion of requested consumer group ('group-0') was successful.
然后我用相同的id ' group -0‘重新创建了这个组。
为什么新成立的集团仍持有100英镑的折价？为什
浏览 2提问于2021-01-27得票数 1
回答已采纳
考虑一下，我有一个三个节点的kafka集群，其中有1000个主题和单个分区。每个主题将被多个消费群体使用，比如100个并行的。因此，总共可以有1000*100个消费群体从卡夫卡平行消费。
我担心的是，这是否会对卡夫卡集群的性能产生更大的影响？
另外，我的模型将为每个用户请求启动使用者，并从队列中获取单个数据，并且在请求结束时消费者将被关闭。对于来自同一个用户的下一个请求，将再次启动同一个组的使用者，并关闭它。这将对所有用户重复。
以这样的方式创造和摧毁消费者是否合适呢？如果没有，是否有其他适当的方法来处理这个用例？
浏览 0提问于2019-04-17得票数 0
我刚接触kafka，所以很难理解pubsub模型。假设我有一个包含2个分区的主题(让我们称它们为partition1，partition2)
我有3个消费者
consumer1、consumer2 -->属于consumerGroup1，每个都绑定到partition1和partition2
consumer3 -->属于consumerGroup2，此group.It中的唯一使用者从partition1读取数据
让我们假设所有3个消费者都从partition1读取了5条消息
我怀疑的是，
1. partition1中的所有5条消息是否都能到达每个消费者，即consumer1和co
浏览 2提问于2021-11-24得票数 0
卡夫卡出口商公开了"kafka_consumergroup_members“标准，它显示了每个消费者群体中有多少成员。然而，我想知道是否有可能列出每个消费群体的实际成员，类似于如何使用“-group-describe-members”选项与卡夫卡-消费者组。be shell脚本输出指定群体的个人消费者。这是否有可能与卡夫卡出口商或任何其他卡夫卡相关的出口商？
浏览 7提问于2022-06-10得票数 0
消费者代码如下：
from kafka.client import KafkaClient
from kafka.consumer import SimpleConsumer
kafka = KafkaClient("localhost", 9092)
consumer = SimpleConsumer(kafka, "my-group", "my-topic")
consumer.seek(0, 2)
for message in consumer:
print message
kafka.close()
然后我用脚本生成消息：
bi
浏览 0提问于2014-02-16得票数 2
回答已采纳
嗨，我正在寻找一个汇合cli命令来使用GetOffsetShell来获取最早和最新的偏移量，并获得消费者组的详细信息。
查找以下kafka命令，的汇合cli命令
获取偏移量详细信息
/opt/kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell \
--broker-list localhost:9092 \
--topic my_topic \
--time -1
消费者组获取列表
kafka-consumer-groups.bat --bootstrap-server localhost:9092 --lis
浏览 9提问于2022-05-05得票数 0
Kafka服务器版本= 0.8.2.2
(Kafka c++ client)版本= 0.9.1
例如，一组中有两个消费者。两个消费者必须从主题中获得不同的消息。
我使用类似于下面，但似乎不起作用。两个消费者从主题中获取所有消息。
std::string topic_str = "sample";
std::string errstr;
int32_t partition = 0;
int64_t start_offset = RdKafka::Topic::OFFSET_END;
RdKafka::Conf *conf = RdKafka::Conf::cre
浏览 1提问于2016-10-05得票数 1
回答已采纳
我正在努力让我的android在MAC上被检测到，并遵循可用的在线指令。打开USB大容量存储导致MAC在Finder中检测到它，但它没有显示在下面的列表中-
adb devices
器件- Micromax A111
Android-4.1.2
USB调试- On
USB存储器-一旦电缆拔出，它就会自动关闭。当它连接起来的时候打开。
adb_usb.ini -添加了供应商编号0x1c9e和做了./adb杀灭服务器
安装了ADB wifi，但前提是让adb识别设备一旦与数据电缆连接，数据电缆就无法工作，并且是根设备。
数据电缆正在使用-一个是随设备一起提供的，也可以插
浏览 8提问于2013-10-13得票数 1
回答已采纳
1回答
我的要求是负载平衡的2 MQTT节点运行在不同的VM上，然后让消费者在两个节点上使用这些MQTT代理。消费者的工作将是订阅一个主题，并在收到数据后，发布给卡夫卡。问题是，如果两个MQTT使用者都订阅了相同的主题，他们将收到相同的消息，并且都会将其插入Kafka，从而创建副本。有没有办法避免把复制品写进卡夫卡？
我已经尝试过Mosquitto和Mosca经纪人，但他们不支持集群。因此，订阅的客户端如果被订阅到不同的节点，那么就不会收到消息，而不是消息发布的节点。这两个节点都在HAProxy后面。
我目前使用的是支持集群的emqtt，它解决了负载平衡问题，但它似乎不支持跨集群节点的共享订阅
浏览 3提问于2017-07-22得票数 1
1回答
我试图监控一个消费群体在Kafka 0.10中的滞后。
我们的消费者正在用卡夫卡而不是ZooKeper跟踪他们的补偿。这意味着我可以使用以下方法获得数字：
bin/kafka-consumer-groups.sh --bootstrap-server <broker> --describe --group <group-name>
这个很好用。但是，我的代理已经使用来收集一些统计数据。我已经将JConsole连接到代理，但看不到kafka-consumer-groups.sh报告的相同数据。
在不需要任何额外的工具的情况下，是否可以通过JMX从Kafka获得这些信息？
浏览 10提问于2017-10-06得票数 2
我有一个spring boot应用程序。我希望实现一个从Kafka主题消费的函数，对于每个消费的消息，它调用另一个方法对消息进行一些处理，然后将其存储在数据库中。源kafka主题有12个分区。我想在同一个消费群体下创建6到12个kafka消费者。所有的消费者都将消费来自Kafka的消息并调用该函数。我不能得到一个好的设计。有人能帮帮我吗？
浏览 1提问于2020-02-19得票数 1
假设我有两个主题(两个分区和无限保留)：
my_topic_amy_topic_b
还有一个消费群体：
my_consumer
在某种程度上，它使用了这两个主题，但是由于一些变化，它对my_topic_a不再感兴趣，所以它停止使用它，现在正在积累延迟：
kafka-consumer-groups.sh --bootstrap-server=kafka.core-kafka.svc.cluster.local:9092 --group my_consumer --describe
TOPIC PARTITION CURRENT-OF
浏览 3提问于2020-12-10得票数 3
回答已采纳
我的问题是，关于卡夫卡：
我有两个程序与卡夫卡交换数据(一个是生产，另一个是阅读)。假设顾客崩溃了。生产者将继续发送消息，一段时间后，我们将重新启动客户。
对于我们当前的代理设置，下面是应该发生的事情：
崩溃发生在不到1天前(因为offsets.retention.minutes是1440)，将检索偏移量，并将处理等待的消息。
事故发生在1天多以前，客户的新冲销将被重新调整为最早的( auto.offset.reset是最早的)。问题是:如果一些消息已经被处理(在1天到7天之间)，它们将再次被处理，因为Kafka保存了7天的消息( (log.retention.hours是168)。
解决方案
浏览 4提问于2017-05-04得票数 0
你好，卡夫卡社区。
我正在开发一个使用kafka streams API的服务。
有没有办法让我的服务检测到它是否与kafka服务器断开了连接？
下面是一些背景知识。我正在尝试为我的服务增加弹性。也就是说，如果我的服务没有连接到kafka，那么它应该发出警报，通知我们有问题。
我正在使用KafkaStreams.setStateListener()安装一个侦听器，该侦听器会在kafka流从RUNNING变为ERROR状态时捕获该侦听器。
虽然这有助于捕获某些错误，但此状态侦听器机制不会捕获其他类型的错误。例如，如果kafka服务器宕机，并且服务不再连接到kafka服务器，则服务将保持RUNNI
浏览 1提问于2019-03-07得票数 1
在我参与的一些使用MKS作为版本控制工具的项目中，这种情况偶尔会发生。文件被检出、修改，然后检入，但忘记了将其包括在相关的变更包中。
有没有办法在事后做到这一点？
谢谢。
浏览 3提问于2013-06-11得票数 0
1回答
我们计划让生产者(Java rest api)和消费者(Java客户端)在多台机器上运行，并且所有消费者属于同一消费者组。
1)在这种情况下，如果我在生产者发布Kafka写入随机/默认分区的消息时没有指定分区...
2)当消费者检索消息时，如果我没有提到分区，我会传递主题名称..and提交偏移量给Kafka服务器。
如果我只有一个消费群体，这是一个好方法吗？
3)我是否需要Kafka rest代理，据我所知这不是必需的，因为我正在使用Java client ..please如果我错了...please请纠正我，让我知道如果我必须在这里使用Kafka rest代理我将获得什么好处...
浏览 11提问于2016-02-24得票数 2
对于某个主题，我有6个分区，从这个主题中有4个消费者在消费。生产者以循环方式对分区进行生产。这4名消费者属于同一消费群体。
通过一些负载测试，我可以看到两个分区的消耗速度非常慢，而其他分区几乎总是空的。我想尽可能增加我的吞吐量。
如果负载在某个时候增加，那么将从kafka默认的分区分配策略是什么？，我希望将我的用户扩展到6(与分区相同的数目，因此它是一个1-1的消费者)。在实现最佳吞吐量的4个消费者场景中，应该限制我的生产者只生产4个分区，直到我增加了我的使用者的数量？。
浏览 5提问于2022-06-13得票数 0
1回答
按需水平缩放事件驱动体系结构
、、、、
当负载增加时，水平扩展事件驱动架构的最佳方法是什么？
许多人建议使用Kakfa作为EDA的消息队列源，但是Kafka只允许每个分区中的一个消费者组中的一个消费者。重新分区，特别是在重负载情况下，可能会花费很大的成本和时间。
在一个消费群体中有许多消费者能够快速地工作和确认，这将带来一些横向扩展，但是现在需要考虑消息顺序以及负载完成。
但是，使用RabbitMQ，可以动态地创建和删除队列，这将需要额外的调度器来帮助管理和分发负载。
也没有解决领域带来的负载平衡问题.
任何帮助都将不胜感激。谢谢
浏览 3提问于2021-02-09得票数 1
我正在阅读Kafka文档，并注意到下面这一行：
但是，请注意，在使用者组中不能有比分区更多的使用者实例。
嗯。我怎么能自动缩放这个？
例如，假设我有一个具有hi/lo优先级的消息传递系统，所以我为消息创建了一个主题，并为hi和lo优先级消息创建了分区。
如果这是RabbitMQ，我将为每个分区分配一组自动可伸缩的使用者，如下所示：
如果我理解卡夫卡模型，在一个消费群体中，每个分区不能有一个消费者，所以卡夫卡的图片不适用于卡夫卡，对吧？
好的，那么像这样的>1个消费群体呢：
是卡夫卡的限制但是..。如果我理解这是如何工作的，那么两个消费者组都将从一个分区(例如m
浏览 12提问于2016-03-24得票数 25
回答已采纳
我需要为所有消费者发送消息，但在检测谁应该收到此消息之前，如何使用Kafka执行此操作？
我是否应该使用Kafks stream来过滤数据，然后发送给消费者？据我所知，每个消费者都应该被添加到唯一的消费群体中，但如何实时检测，谁必须收到消息？
浏览 19提问于2020-05-13得票数 2
回答已采纳
环境
3节点卡夫卡集群
亚马逊MSK
v2.3
1主题
6个分区
1个消费群，2个消费者
在Kubernetes跑
合流.NET SDK 1.2.2
除了bootstrap.servers和group.id之外，所有默认设置。
问题
首先，我的一个消费者遇到了以下异常。
Confluent.Kafka.KafkaException: Broker: Specified group generation id is not valid
at Confluent.Kafka.Impl.SafeKafkaHandle
浏览 2提问于2020-10-30得票数 5
3回答
我正在使用kafka-node ConsumerGroup来阅读卡夫卡的信息。我想从主题中读取新消息，即不是所有以前未使用的消息，而是在创建ConsumerGroup之后到达的消息。
只有当给定的fromOffset: 'latest'没有存储偏移量时，选项groupId才会出现。我可以通过在每次创建groupId时创建一个新的ConsumerGroup来使用这个方法，但是我希望避免这种方法，因为这会在ConsumerGroup服务器上创建大量的groupIds和存储偏移量。
我还尝试手动将偏移量设置为-1，如下所示：
offset = new kafka.Offset(con
浏览 2提问于2017-08-25得票数 3
回答已采纳
我已经通过以下代码将G1组中一个Kafka消费者的fetch偏移量倒带到了off1：
consumer1.set_topic_partitions((topic, partition, off1))
在上面一行中，off1是对应分区中该主题的最旧可用消息的偏移量。
现在我正在实例化一个不同群组G2中的Kafka消费者，如下所示：
consumer2 = KafkaConsumer(bootstrap_servers=broker,
auto_offset_reset='smallest',
浏览 1提问于2015-09-03得票数 1
1回答
在Kafka中，我可以将我的主题分成许多分区。在Kafka中，我不能拥有比分区更多的使用者，因为分区被用作扩展主题的一种方式。如果我有更多的负载，我可以增加分区的数量，这将允许我增加使用者的数量，这将允许我对一个给定的主题进行更多的线程/进程处理。
在卡夫卡，有一个消费者集团的概念。如果我们有10个消费者组在一个单一的主题，每个消费者群体将有机会处理每一个主题中的每一个信息。使用者组仍然利用来自分区的可伸缩性(即每个使用者组最多可以拥有'n‘使用者，其中'n’是一个主题上的分区数)。这就是卡夫卡的美，可扩展性和多通道阅读是两个独立的概念，两个独立的旋钮要转动。
在Kinesis
浏览 0提问于2018-05-05得票数 7
回答已采纳
4回答
我在本地机器上安装了Kafka，并启动了动物园管理员和一个代理服务器。
现在我有一个主题，描述如下：
~/Documents/backups/kafka_2.12-2.2.0/data/kafka$ kafka-topics.sh --zookeeper 127.0.0.1:2181 --topic edu-topic --describe
Topic:edu-topic PartitionCount:3 ReplicationFactor:1 Configs:
Topic: edu-topic Partition: 0 Leader: 0 Replicas: 0
浏览 3提问于2019-11-19得票数 5
回答已采纳
1回答
我正在设法获得卡夫卡集群的当前使用统计数据。我希望收集以下资料：
kafka集群中的主题数
每个kafka代理的分区数
活跃消费者和生产者的数目
每个kafka代理的客户连接数
每个分区上的消息数量、磁盘大小等。
滞后复制、消费滞后等。
积极消费群体
任何其他可以和应该收集的统计数据，目前我正在考虑收集上面的统计数据。
我可以得到1和2使用动物园管理员的公用设施，但我是迷失在休息。我看过Jconsole中的mbean，但没有找到上面的任何内容。我还尝试使用基于regex的表达式来获取这些mbean，但这也不起作用。
我使用的是Kafka v2.1和新的消费a
浏览 1提问于2019-02-12得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
