如何让Kafka消费者发现不存在的主题，并从元数据中删除它们？
在Kafka中，消费者无法直接发现不存在的主题并从元数据中删除它们。Kafka的元数据存储在Zookeeper中，而消费者只能通过Zookeeper获取元数据信息。因此，要实现消费者发现不存在的主题并删除它们，需要进行以下步骤：
- 连接到Zookeeper：使用Kafka提供的Zookeeper客户端库，连接到Zookeeper服务器。
- 获取所有主题列表：通过Zookeeper客户端，获取Kafka集群中的所有主题列表。
- 获取消费者组的消费者偏移量：对于每个消费者组，通过Zookeeper客户端获取其消费者偏移量。
- 检查主题是否存在：对于每个主题，检查其是否存在于主题列表中。
- 删除不存在的主题：如果发现某个主题不存在于主题列表中，使用Zookeeper客户端删除该主题的消费者偏移量。
需要注意的是，这种方法只能删除消费者组的消费者偏移量，而不能直接删除Kafka集群中的主题。如果需要删除主题，可以使用Kafka提供的工具或API进行操作。
推荐的腾讯云相关产品：腾讯云消息队列 CMQ、腾讯云云原生数据库 TDSQL、腾讯云云服务器 CVM、腾讯云云安全中心 SSC、腾讯云云点播 VOD、腾讯云人工智能 AI Lab、腾讯云物联网 IoV、腾讯云移动开发 MSDK、腾讯云云存储 COS、腾讯云区块链 TBaaS、腾讯云腾讯元宇宙 TME。
更多产品介绍和详细信息，请访问腾讯云官方网站：https://cloud.tencent.com/
相关·内容
1回答
如何让Kafka消费者发现不存在的主题，并从元数据中删除它们？
apache-kafka
我的KafkaConsumer订阅了一个主题模式。但是，当删除一些过时的主题时，由于消费者客户端仍然在其主题列表中维护已删除的主题，所以从集群中获取元数据时会遇到UNKNOWN_TOPIC_OR_PARTITION异常。这些过时的主题信息在消费者中永远不会过期，所以我必须重新
浏览 54提问于2019-04-24得票数 0
3回答
kafka中的Bootstrap server vs zookeeper？
apache-kafka、apache-zookeeper
为什么不推荐使用kafka-consumer中的zookeeper，为什么建议使用bootstrap服务器？bootstrap-server有什么优点？
浏览 0提问于2017-09-12得票数 36
回答已采纳
1回答
Kafka主题创建后无法分配领导者
apache-kafka
我使用的是kafka主题，它在我的应用程序中也是元数据。我通过删除与该主题对应的目录，从zookeeper shell中硬删除了该主题。再次创建主题后，我对该主题进行了描述，并发现没有为这个新创建的主题分配领导者。在消费者中，我可以看到重复
浏览 3提问于2018-12-03得票数 0
2回答
Kafka -有可能改变集群大小，同时保持改变对生产者和消费者透明吗？
apache-kafka
我正在对Kafka进行调查，以评估它是否适合我们的用例。你能帮我理解一下Kafka在改变现有集群大小时有多灵活吗？
我正在对Kafka进行调查，以评估它是否适合我们的用例。你能帮我理解一下Kafka在不拆掉集群的情况下向现有集群添加brokers的灵活性吗？这样做的时候有什么需要注意的吗？
浏览 2提问于2019-01-08得票数 0
1回答
卡夫卡+消费集团的根本原因是再平衡
apache-kafka、kafka-consumer-api
卡夫卡机器是作为hortonworks软件包的一部分安装的，kafka版本是0.1X。/usr/hdp/current/kafka-broker/bin/kafka-consumer-groups.sh --group deeg_dat
浏览 2提问于2021-12-23得票数 2
回答已采纳
1回答
Zookeeper客户端无法rmr /brokers/TOPIC/MY_TOPIC
apache-kafka、apache-zookeeper
我正在尝试删除一个包含8个分区和2个副本的Kafka主题。首先，我使用kafka-topic.sh --delete命令删除该主题。然而，我仍然在/brokers/ topic /中看到了这个主题。我试着重启Kafka，一切照旧。
顺便说一句，可以成功删除1个分区和1个副本的主题。
浏览 22提问于2017-03-16得票数 0
回答已采纳
1回答
kafka是否支持代理端的通配符主题匹配和订阅？
apache-kafka
希望消费者订阅所有的主题I*卡夫卡集群/代理是否可以自动为消费者订阅以i开头的新主题？做这种动态主题过滤/发现/订阅的唯一方法是由使用者每次调用元数据刷新并检查添加的新主题</e
浏览 1提问于2019-02-04得票数 0
回答已采纳
1回答
Flum1.7& Kafka -如何在话题开始时重新启动？
apache-kafka、flume
我使用Flumem1.7Kafka source将Apache中的数据提取到我的AbstractSink中。在过去，我可以通过使用./kafka-consumer-groups.sh --delete删除主题偏移，在主题开始时重新启动偏移量，但是由于Flum1.7(显然)使用了“新”使用者，所以尝试./kafka-consumer-groups.sh --delete现在会给出以
浏览 2提问于2017-01-18得票数 0
回答已采纳
2回答
如何通知消费者在Kafka中创建了新的topic？
apache-kafka、sarama
我正在尝试让我的消费者动态更新它的消耗。
让我给你一个使用动物的更具体的例子。假设我有一家宠物店，每个主题都是一种动物(例如狗、猫、鱼)。我的Kafka消费者的主要责任是抓取我们在Kafka中的任何日志/记录/消息，并将它们存储到数据库中。假设我的
浏览 2提问于2017-10-11得票数 0
2回答
kafka连接器中的动态主题
apache-kafka、apache-kafka-connect、confluent-platform
kafka增加了在连接器中使用regex的新功能，但是似乎连接器启动后新添加的主题数据直到连接器重新启动时才会被消耗。我们需要动态添加新的主题，并让连接器根据连接器属性中定义的正则表达式使用该主题。例如: regex: topic-.* topic: topic-1，topic-2如果我引入了新的topic topic-3，那么如何让连接器在不重
浏览 11提问于2018-07-31得票数 3
2回答
是否有卡夫卡生产者API吐露设置主题自动创建？
apache-kafka、apache-kafka-connect、kafka-producer-api
Kafka有自己的一组配置，可以以编程方式创建主题。
对于生产者(java) API?有类似的东西吗？
浏览 7提问于2022-06-22得票数 2
回答已采纳
1回答
在卡夫卡不存在的起始位置，如何向制片人索取数据？
apache-kafka、producer-consumer
我有一个带有时间序列数据的数据库，这些数据被发送到Kafka。例如，如果我请求一个卡夫卡中不存在的偏移量
浏览 6提问于2022-04-10得票数 0
回答已采纳
2回答
如何获取Kafka消费者的偏移量？
apache-kafka、spark-streaming、kafka-consumer-api、kafka-producer-api
使用Kafka(v2.11-0.10.1.0)-spark-streaming(v-2.0.1-bin-hadoop2.7).
我有Kafka生产者和Spark-streaming消费者来生产和消费。一切正常，直到我停止消费者(大约2分钟)并重新开始。消费者启动并读取数据，绝对完美。但是，我迷失了2分钟的数据，其中消费者是关闭的。Kafka consumer/server.properti
浏览 16提问于2017-01-18得票数 0
4回答
每个消费者订阅Kafka主题的最大限制
apache-kafka、kafka-consumer-api
在Kafka中，一个消费者可以订阅的主题的最大限制是什么。我在任何地方都找不到这个记录的值。
如果消费者订阅了500000个或更多的主题，性能是否会下降。
浏览 6提问于2017-09-19得票数 5
2回答
如何删除kafka消费者群体(通过新的消费api创建)？
java、apache-kafka、kafka-consumer-api
我通过新的消费API创造了卡夫卡消费者。我们有一个使用者组，每个组中有一个使用者实例。Kafka脚本“kafka- old groups.sh”提供了删除用户的方法，但这只适用于运行命令的旧使用者组：给出
--删除：警告:组删除只适用于旧的
浏览 2提问于2016-06-10得票数 7
1回答
使用kafka- Consumer groups.sh脚本删除消费者组是删除消费者组元数据的安全方法吗？
apache-kafka、kafka-consumer-api
我使用多线程kafka使用者作为一个单独的60个分区kafka主题，每个使用者线程映射到每个分区。在消费完成后，使用KafkaConsumer.close()方法关闭每个使用者线程。在同一消费者不满意的情况下进行再消费期间(寻求开始是每个消费者的步骤)，1或2个使用者线程无法从它们分配的分区中消耗。如果提供了新的使用者组id，则不会出现此问题。因此，如果使用下面的脚本删除<
浏览 2提问于2019-10-25得票数 0
1回答
当服务器重新启动后未使用消费者group.id时
apache-kafka
当消费者实例group.id在重启时更改时，是否会对kafka性能产生任何影响。较旧的group.id将会发生什么情况?它是否仍然存在于代理内存中，或者何时将被删除？假设我有1000+消费者实例，并且所有实例都在重启时动态分配group.id。
可以为{log.retention.ms'}提供什么列表值。可以设置为1ms吗？
浏览 8提问于2018-08-28得票数 0
回答已采纳
3回答
流处理后从kafka主题中删除消息
apache-kafka、apache-kafka-streams
在我们的事务处理系统中实现了kafka流处理。该解决方案实现如下：
Kafka producer将事件发布到kafka topic，流处理器处理输入事件并进行聚合操作。流处理完成后，该事件将发布到另一个主题。由于第一个topic中没有实现消费者，如何从第一个topic中移除已处理的消息。
浏览 0提问于2017-08-14得票数 1
2回答
__consumer_offsets如何为两个消费者保留偏移量？
apache-kafka
就像上图一样，如果消费者A从一个分区消费，而消费者B从同一个分区消费，那么Kafka如何管理__consumer_offsets中的偏移量
我想知道卡夫卡是如何在.index，.log，.timeindex文件中写入消费者偏移量的。
浏览 37提问于2019-07-09得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
即时通信 IM
ICP备案
对象存储
实时音视频活动推荐
腾讯云开发者
