kafka-consumer-api 相关问题

在控制台模式下运行时，Kafka的并行性概念是否适用？

所以，我是Kafka的新手，并且已经阅读了一段时间。我发现有关融合的信息。 https://docs.confluent.io/current/streams/architecture.html所以我从中了解到...

[Kafka消费邮件时使用汉克

我有使用来自生产者的消息的以下代码片段：公共类KafkaAvroExampleWithWait：MonoBehaviour {ConsumeResult 消耗Result2;字符串...

Kafka消费者组重新平衡

我正在使用kafka用户组管理来处理我的消息。我的邮件的处理时间互不相同。因此，对于最大记录数20，我将最大轮询间隔设置为20分钟。...

手动确认Kafka Batch监听器。多线程我做错了什么？

我正在创建一个应用程序，该应用程序使用来自kafka的数据并对其进行一些处理。由于此应用程序的工作范围很大，因此我正在使用并发批处理侦听器（max.poll ....

如何为具有多个分区的Kafka主题启动多个使用者？

我最近开始使用spark，并且不得不处理在spark中使用Kafka主题的多个分区的情况。如何启动多个消费者？我是否需要多个...

从Kafka获取具有特定键的所有事件

我的系统中有一些主题，可以存储给定实体的事件。现在，我想对事件日志进行一些分析。因此，我需要查询属于...

我可以从Kafka的被动段中读取数据吗？

假设对于一个分区，我有这样的段； Segment-0->起始偏移量：0，结束偏移量：1087 Segment-1->起始偏移量：1088结束偏移量：2300 Segment-2（活动段）->起始偏移量：2301，......

Kafka使用者：如何在Python中读取特定的Avro字段？

在使用者的以下摘要中，我能够接收发送的数据。我如何从整个数据中访问特定值以进行处理。从confluent_kafka导入KafkaError，从...

Kafka消费者多租户

我是写Kafka使用者的新手，如果我有两个使用者在同一个组ID下运行，并且有两个分区，我会遇到这种情况。假设使用者1 ===>链接到====>分区1 ...

如果处理失败，再次使用同一条消息

我正在使用Confluent.Kafka .NET客户端版本1.3.0。我正在关注文档：var ConsumerConfig = new ConsumerConfig {BootstrapServers =“ server1，server2”，AutoOffsetReset = ...

Kafka-如果消息处理失败，如何再次使用同一条消息

我正在使用Confluent.Kafka .NET客户端版本1.3.0。我正在关注文档：var ConsumerConfig = new ConsumerConfig {BootstrapServers =“ server1，server2”，AutoOffsetReset = ...

墓碑和密钥库清理

我有一些用Java实现的Kafka使用者，我正在实现一个独立的应用程序来检查记录并对其进行逻辑删除。希望卡夫卡在压缩时会删除状态存储...

如何使用Java从Kafka消费消息，从特定的偏移量开始

最早阅读：props.put（ConsumerConfig.AUTO_OFFSET_RESET_CONFIG，“最早”）；从最新版本中读取：props.put（ConsumerConfig.AUTO_OFFSET_RESET_CONFIG，“最新”）；但是哪一行...

[使用Kafka侦听器vs Kafka使用者获取具有指定密钥的消息

我使用SpringBoot应用程序生成或使用/收听Kafka消息。我在主题中产生一条消息，并通过比较messageKey来使用/收听特定消息，然后发送...

独立消费者抛出InvalidGroupIdException

这是一个两部分的问题。 I.代码是：import java.time.Duration import java.util import java.util.Properties import org.apache.kafka.clients.consumer。{ConsumerConfig，KafkaConsumer} import ...

从Kafka主题中的偏移量之间读取消息

我正在寻找一种方法来消耗我的Kafka主题中具有特定偏移范围的某些消息（假设我的分区的偏移量为200-300，我想使用250-270偏移量的消息...

来自不同消费者组的多个消费者如何从同一个分区读取？

我有一个用例，其中我有2个消费者位于不同的消费者组（cg1和cg2）中，订阅了具有4个分区的同一主题（主题A）。如果两个使用者都从相同的位置读取内容，会发生什么情况？

如何在Apache Kafka中查看主题属性

我想查看主题级别的属性，例如“ message.timestamp.type”：“ LogAppendTime”，“ cleanup.policy”：“ compact”，是否可以查看在主题级别设置的所有属性？有...

[kafka控制台使用者漂亮的打印json

如何打印从Kafka控制台使用者到JSON漂亮的响应。我已经尝试过了，但是没有用。 ./kafka-console-consumer.sh --bootstrap-server localhost：9092 --topic indexing-v1 --from -...

如何动态获取主题名称，然后从中读取

我有一个需要动态读取主题名称的场景。有什么方法可以获取域中所有主题的名称，然后逐一读取它们的消息？ @KafkaListener（groupId =“ ...