kafka-consumer-api相关内容

消费者组成员没有分区

我在同一个消费者组上启动了两个消费者，我订阅了 20 个主题(每个主题只有一个分区) 仅在消费者上使用: kafka-consumer-groups --bootstrap-server XXXXX:9092 --group foo--describe --members --verbose 注意:这不会显示有关旧的基于 Zookeeper 的消费者的信息.消费者 ID 主机客户端 ..

发布时间：2021-11-12 03:20:57 apache-kafka kafka-consumer-api 其他开发

如何在Java中获得消费者Kafka滞后

我在 java 中有一个生产者，在 nodeJS 中有一个消费者.我想在 Java 中知道什么是消费者滞后，所以我知道我是否可以为该主题生成更多数据. java 中获取消费者延迟的 API 是什么? 解决方案 您可以从 Java 调用的实际类是 kafka.admin.ConsumerGroupCommand.它是 Scala 代码，但很容易直接从 Java 调用. 这里有一个 ..

发布时间：2021-11-12 03:20:54 java apache-kafka kafka-consumer-api lag producer-consumer Java开发

无法在 Spring Boot 中使用 Kafka 消息

我们有一个使用 org.apache.kafka.clients.consumer.KafkaConsumer 消费 Kafka 消息的 Java 应用程序 我们创建了一个具有 Spring-Kafka 依赖项的 Spring Boot 应用程序，但无法读取新项目中的消息.检查了明显的参数，包括引导服务器的主机名和端口(日志显示可以识别)、组、主题以及 Spring Boot 与原始使用者一样 ..

发布时间：2021-11-12 03:20:21 spring-boot apache-kafka kafka-consumer-api spring-kafka 其他开发

Confluent 控制中心拦截器

如何将 Confluent Control Center Interceptor 添加到现有的 S3(Sink) Connector?监视接收器.我正在寻找文档.任何帮助表示赞赏. 解决方案 要绝对清楚，您的接收器上需要拦截器和源.如果不这样做，您将无法使用 Confluent Control Center 监控您的管道，就像现在一样. 要在 Kafka Connect 中启用拦截器 ..

发布时间：2021-11-12 03:20:18 apache-kafka kafka-consumer-api apache-kafka-connect confluent-platform 其他开发

Kafka + Spring Batch 监听器刷新批处理

使用 Kafka 代理:1.0.1spring-kafka: 2.1.6.RELEASE 我正在使用具有以下设置的批量消费者: //其他设置不显示..props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "100"); 我以如下方式使用 spring 监听器: @KafkaListener(topics = "${topics}", gr ..

发布时间：2021-11-12 03:20:15 apache-kafka kafka-consumer-api spring-kafka 其他开发

如果我在 Kafka 中有事务性生产者，我可以使用 Kafka Streams 读取一次消息吗?

我想拥有 Exactly-once 语义，但我不想使用 Consumer 读取消息.我宁愿使用 Kafka Streams AP 阅读消息.如果我将 processing.guarantee=exactly_once 添加到 Stream 配置中，是否会保留恰好一次语义? 解决方案 Exactly-once 处理基于读-处理-写模式.Kafka Streams 使用这种模式，因此，如果您编 ..

发布时间：2021-11-12 03:19:54 apache-kafka kafka-consumer-api kafka-producer-api apache-kafka-streams 其他开发

kafka 偏移值是如何计算的?

kafka 偏移值是否每个分区或每个主题唯一(考虑相同的组 ID)? 解决方案 每个分区是唯一的.从零和长数据类型开始. ..

发布时间：2021-11-12 03:19:33 apache-kafka kafka-consumer-api 其他开发

为什么将元数据添加到此 Kafka 连接器的输出中?

我有一个 Kafka 连接器，其中包含 SourceTask 实现中 poll() 方法的以下代码. @Override公共列表poll() 抛出 InterruptedException{SomeType item = mQueue.take();列表记录 = 新的 ArrayList();SourceRecord[] sourceRecords = 新 SourceRecord[]{新的 S ..

发布时间：2021-11-12 03:19:24 java apache-kafka kafka-consumer-api apache-kafka-connect Java开发

卡夫卡消费者 - 不一致地接收消息

我可以在命令行上针对 Kafka 位置安装发送和接收消息.我还可以通过 Java 代码发送消息.这些消息显示在 Kafka 命令提示符中.我还有一个供 Kafka 消费者使用的 Java 代码.代码昨天收到消息.但是，今天早上它没有收到任何消息.代码没有改变.我想知道属性配置是否不太正确.这是我的配置: 制作人: bootstrap.servers - 本地主机:9092group.id ..

发布时间：2021-11-12 03:19:13 apache-kafka kafka-consumer-api 其他开发

使用主题启用 ACL 的 Kafka java 生产者和消费者

我对 kafka ACL 配置有点困惑，我们为生产者和消费者配置授权.有各种示例显示使用命令行生成/使用消息.我们是否需要任何额外的配置来使用 JAVA api 向/从安全的 kafka 主题生成/使用消息. 解决方案 如果您想了解安全 Kafka 服务器的配置细节，Confluent docs 有很好的描述.您可以在文档中注意到您通过文件 client.properties 设置某些属性 ..

发布时间：2021-11-12 03:19:04 java apache-kafka kafka-consumer-api kafka-producer-api Java开发

如果 Kafka Consumer 实例死亡会发生什么?

Kafka Broker 有 3 个分区. Kafka 消费者实例的数量为 3. 突然，一个消费者实例死亡. 我知道如果 Kafka Consumer 实例死亡，Kafka Broker 正在重新平衡，另一个消费者实例会被分配到该分区. 我想知道假设另一个实例消耗了它最初消耗的所有分区然后分配和消耗死分区是否正确. (我是否必须在客户端代码中实现 ConsumerRebala ..

发布时间：2021-11-12 03:18:47 apache-kafka kafka-consumer-api partition 其他开发

Kafka消费者api订阅主题失败

我正在使用简单的 Kafka 客户端 API.据我所知，消费者消息有两种方式，订阅主题和为消费者分配分区. 但是第一种方法不起作用.消费者 poll() 将永远挂起.它仅适用于 assign. //消费者的通用配置映射config = new HashMap();config.put("bootstrap.servers", bootstrap);config.put("group.id" ..

发布时间：2021-11-12 03:18:26 java apache-kafka kafka-consumer-api Java开发

Kafka 消费者组再平衡

我正在使用 kafka 消费者组管理来处理我的消息. 我的消息的处理时间各不相同.所以我将最大轮询间隔设置为 20 分钟，最大记录数为 20.并且我使用了 5 个分区和 5 个消费者实例，除了上述两个之外，还有默认配置值. 但我仍然间歇性地收到以下错误: [Consumer clientId=consumer-3, groupId=amc_dashboard_analytics] 尝 ..

发布时间：2021-11-12 03:17:52 apache-kafka kafka-consumer-api 其他开发

如何将消息从单个生产者发布到两个主题并在 Kafka 中的单个侦听器中消费

我正在尝试从单个生产者发布两个不同主题的消息. 这里我创建了两个主题: @Bean公共新话题 multi1() {返回 TopicBuilder.name("multi1").partitions(1).build();}@豆角，扁豆公共新话题 multi2() {返回 TopicBuilder.name("multi2").partitions(1).build();} 这是我向两个 ..

发布时间：2021-11-12 03:17:13 spring spring-boot apache-kafka kafka-consumer-api spring-kafka 其他开发

无法从 kafka 主题轮询/获取所有记录

我正在尝试从特定主题中轮询数据，例如 kafka 正在接收 100 条记录/秒但大多数时候它不会获取所有记录.我使用超时为 5000 毫秒，我每隔 100ms 调用此方法注意:我也订阅了特定主题 @Scheduled(fixedDelayString = "100") public void pollRecords() {ConsumerRecords记录 =LeadConsumer.p ..

发布时间：2021-11-12 03:17:04 java apache-kafka kafka-consumer-api spring-kafka Java开发

卡夫卡未提交的消息不再被消费

我正在处理 kafka 消息并使用 Spark 流插入 kudu 表，手动偏移提交这里是我的代码. val topicsSet =topics.split(",").toSetval kafkaParams = Map[String, Object](ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG ->经纪人，ConsumerConfig.GROUP_ID_CON ..

发布时间：2021-11-12 03:16:41 apache-kafka spark-streaming kafka-consumer-api 其他开发

Kafka Connect SMT 添加 Kafka 标头字段

我需要找到或编写一个将向请求添加标头字段的 SMT.请求缺少一些类型字段，我想添加它们. 您究竟如何在 SMT 中添加标题，我所看到的只是如下所示的记录转换，但如果我想更改标题或向其中添加字段怎么办? 私有 R applySchemaless(R record) {最终映射value = requireMap(operatingValue(record), 目的);//record.he ..

发布时间：2021-11-12 03:15:48 apache-kafka kafka-consumer-api apache-kafka-connect 其他开发

Kafka 消费者组再平衡

我正在使用 kafka 消费者组管理来处理我的消息. 我的消息的处理时间各不相同.所以我将最大轮询间隔设置为 20 分钟，最大记录数为 20.并且我使用了 5 个分区和 5 个消费者实例，除了上述两个之外，还有默认配置值. 但我仍然间歇性地收到以下错误: [Consumer clientId=consumer-3, groupId=amc_dashboard_analytics] 尝 ..

发布时间：2021-11-12 03:15:45 apache-kafka kafka-consumer-api 其他开发

Python Kafka消费者从一开始就没有收到消息?

我在 Windows PC 上安装了 Kafka.创建了一个主题 quickstart-events 并发送了一些消息.运行带有参数 --from-beginning 的控制台使用者可以接收消息. .\bin\windows\kafka-console-consumer.bat --topic quickstart-events --from-beginning --bootstrap-serv ..

发布时间：2021-11-12 03:15:10 python apache-kafka kafka-consumer-api Python

Spring Kafka:在运行时订阅新的主题模式

我正在使用注解 @KafkaListener 在我的应用程序中使用主题.我需要在运行时更改已经运行的使用者的主题模式，以便可以使用与新模式匹配的新主题. 我尝试了下面的代码，但它仍然使用与旧主题模式匹配的主题.在这里，我在应用程序启动时设置了“旧主题模式".然后，我使用 Spring @Scheduler 每 10 秒将模式更新为“new-topic-pattern". 类“KafkaTo ..

发布时间：2021-11-12 03:14:33 java spring apache-kafka kafka-consumer-api spring-kafka Java开发