kafka-consumer-api 相关问题

有没有一种方法可以使用 Spring Kafka 从给定的 kafka 主题中检索消费者延迟？

我正在使用 Spring Kafka 并有一个监听器使用来自多个主题的消息，定义为： @KafkaListener(话题={ “c1.ERROR”，“c2.ERROR”，“c3.ERROR”}，groupId ...

Kafka Consumer InstanceAlreadyExistsException

我正在尝试创建一个订阅主题并定期轮询的kafka消费者。在@Bean 下面创建 kafkaconsumer，它由执行程序服务在 @postconstruct 中使用。在其他用途...

Kafka Consumer 消费消息时间长

我正在使用 Spring Kafka 发布消息。当我向 kafka 主题发布 100 条消息时，我可以看到它被 Kafka 消费者接收，延迟至少 1 秒甚至 4 到 5 秒

使用 assign() 确定 kafka 消费者准备就绪的正确方法是什么？

我读过这个问题还有这个。但是我找不到任何关于如何在使用 assign() 启动消费者时确定消费者准备就绪的讨论 现在我的代码看起来像 b...

当一个代理服务关闭时，卡夫卡消费的异常

当复制因子设置为 3 时，我们有 3 个节点的 Kafka 生产集群 当 Kafka 代理服务之一关闭时，我们注意到以下问题（PIC 1） 让我们说 Kafka broker02 我...

订阅同一主题时，免费消费者的行为如何？

本答案中讨论的免费消费者是没有 group.id 的消费者。他们不订阅主题，但主题分区分配给他们。 有人可以让我知道免费消费是如何......

如何在 Kafka 中处理错误的主题名称

我正在尝试使用 Kafka 消费者对象从错误/不存在的主题名称中消费。 下面是我在投票行之后的日志。 警告 19508 --- [广告 | producer-1] org.apa...

KafkaConsumer close() 抛出 ConcurrentModificationException [关闭]

private final CopyOnWriteArrayList openedResources = new CopyOnWriteArrayList<>(); final KafkaConsumer consumer = new KafkaConsumer<>(clientConfig...

Kafka CooperativeStickyAssignor 将所有分区分配给 1 个消费者

设置： kafka 2.8（服务器）和 java org.apache.kafka kafka-clients 版本：3.0.0 运行 kafka 消费者的 kubernetes 上的 4 个 pod 消费者属性：最大轮询间隔为 5 分钟，会话为 45 秒...

设计 Apache-Kafka 场景

我有以下场景： 我正在使用 Apache-Kafka 开发我的纳税和管理税收应用程序。 我有一个主要用例，我认为需要通过 Apache 实施

Kafka Receiver 在 Kafka 重启后无法消费来自主题的消息

最初，Kafka 接收器按预期工作，但是一旦 Kafka 代理重新启动并返回，同一接收器将无法接收消息，并且除了警告之外，日志中没有看到任何错误...

在类级别使用@KafkaListener 时出错

我正在尝试在类级别使用@KafkaListener，并在参数中使用@KafkaHandler 和@Payload 来验证消息，我需要这个，因为主题有不止一种类型的Json。但是我有...

Kafka Consumer 没有收到来自主题的消息——如何调试？

Producer.py 从 binance.websocket.spot.websocket_client 导入 SpotWebsocketClient 从 binance.spot 导入 Spot 作为客户端 从 kafka.admin 导入 KafkaAdminClient，NewTopic 从卡夫卡导入

Kafka 消费者记录 - 处理

对于 kafka，我使用 getmany 来读取消费者消息。在总共 650 条消息中（大约需要 3 天的时间来处理），处理大约需要 100-150 条记录（有时需要 12 小时或

Kafka 自定义反序列化器转换为 Java 对象

我正在使用 Spring Kafka 集成，我有自己的值通用序列化器/反序列化器，如下所示 序列化器： 公共类 KafkaSerializer 实现 Serializer { 初...

Spring-Kafka如何处理多个Kafka Listener和多个Dead Letter Queue？

我正在做一个 Spring Boot 2.7.9 项目。该项目有 3 个 Kafka 监听器，配置如下： @配置 @EnableKafka 公共类 ConsumerKafkaConfig { 私人最终 KafkaProperties

具有相同组 id 的两个 Kafka 控制台消费者不分担负载

使用以下命令启动 Zookeeper ./zookeeper-server-start.sh ../config/zookeeper.properties 启动 Kafka 代理 ./kafka-server-start.sh ../config/server.properties 创建主题...

Kafka 批处理监听器错误的反序列化消息

我正在使用以下配置进行批量侦听，但我的消息错误地反序列化： @KafkaListener( id = "${kafka.buyers.product-sales-pricing.id}", topics = "${kafka.buyers.product-

如何使用 C# 中的 Kafka Confluent Consumer 使用来自 Kafka 主题的最后一条（最新）消息？

我有以下代码用于从 Kafka 主题读取数据。我的目标是定期阅读主题中的最新消息，因为我想在实时图表中使用数据。我写了

kafka逐渐增加生产者和消费者的内存使用问题

我面临 Kafka 生产者和消费者的内存问题，我们的应用程序部署在 OpenShift 上。生产者和消费者应用程序的内存不断增加。 加载 43200...