apache-kafka-streams 相关问题

kafka 中的分区不均匀/无密钥

我有一个主题，有 3 个分区，只有 1 个使用者，我使用的是默认分区器，在本例中为“粘性”。其他一切都默认。 从生产者发送的数据...

Kafka Streams 内存使用率出乎意料地高

我正在 Kubernetes 中运行 Kafka Streams 应用程序，该应用程序只需将主题直接读取到 GlobalKTable 中，然后提供 HTTP API 来获取键值对。该 Pod 的内存使用量是...

从不同的应用程序访问全局ktable或globalstore - kstream

我们有2个微服务： 负责消费消息并使用配置信息更新 globalKtable (key=id, value=myObject) 一旦使用某个 id 作为输入激活它，它应该......

使用 Kafka 流/KSQL 连接表？

我正在导入一个数据库，其中包含一些表示多对多和一对多关系的链接表。 现在让我们关注一对多关系。例如。一个 Biossay 可以有很多文档...

Kafka Streams - Kafka 1.0.0 中的重新平衡异常

在 Kafka Streams 1.0.0 中，我们看到一个奇怪的错误。我的流应用程序摄取一个 kafka 主题并在不同的状态存储上发出多个聚合，现在该应用程序在 2 个节点的集群上运行，...

kafka 流全局存储在插入的键上返回 null

所以我有一个流写入全局状态存储并从中读取，并且在很多时候它最终是一致的 我已经尝试过他的顺序： 1.) 向流发送新消息 2.) 从

KStream-当第二个匹配值不存在时，KStream leftjoin 不会产生任何值

我们有这些 POJO： @数据 @NoArgsConstruct @AllArgsConstruct 类我的帖子{ 私有字符串内容； 私人 SeenInfo seeInfo; } @数据 @NoArgsConstruct @AllArgsConstruct 班级

为什么 Kafka Streams 创建用于聚合和连接的主题

我最近创建了我的第一个 Kafka 流应用程序用于学习。我使用了 spring-cloud-stream-kafka-binding。这是一个简单的电子商务系统，我正在其中阅读一个名为产品的主题，其中...

带有Spring云流Binder和Json反序列化器的模式注册表

我是 Kafka 的新手，我正在尝试使用模式注册表并与 spring 云流绑定消费者集成，我已经研究了很多并获得了很多参考资料，但没有真正解决问题。所以可以...

基于 Quarkus 的 Kafka Streams 应用程序中的多个拓扑

我正在尝试使用 Quarkus 构建 Kafka Streams 应用程序。我想知道如何在同一个 Quarkus 应用程序中拥有多个拓扑？ 我试图像这样实现它： @ApplicationScoped 公开课

Kafka Streams：如何确保处理完成后提交偏移量

我想使用 Kafka 流处理 Kafka 主题中存在的消息。 处理的最后一步是将结果放入数据库表中。避免数据库争用相关问题（

如何使用Kafka Stream手动提交？

有没有办法使用 Kafka Stream 手动提交？ 通常使用 KafkaConsumer 时，我会执行如下操作： 而（真）{ ConsumerRecords 记录 = Consumer.pol...

'store(org.apache.kafka.streams.StoreQueryParameters<T>)'不能应用于'(java.lang.String, org.apac

我正在使用流表示例，如下所示：https://udemy.com/course/kafka-streams-real-time-stream-processing-master-class/learn/lecture/14244016#questions，当我升级kafka依赖...

如何处理内部 kafka-streams 主题的 avro 架构？

我们正在为代表“公共”接口的 kafka 主题使用 Avro 方案。当这样的主题现在在 Kafka-Streams 应用程序中使用并且调用有状态操作时，

Kafka Streams可以从s3读取数据并将其发送到其他目的地，例如mysql或olap吗？

有一些关于 kafka 流的文章，它从 s3 读取数据并将其发送到其他目的地，例如 olap 或 oltp。 我看不到它们的示例代码，所以我是否更正 kafka

向 Spring Kafka 消费者指标添加自定义静态标签

我的应用程序使用来自 2 个不同集群的相同主题名称 唯一出现的指标标签是“spring.id”， 需要添加更多标签 - 例如集群名称 试图实现这个 哈...

将自定义类存储到 TimestampedKeyValueStore

我有点挣扎。我将来自 Kafka 的消息存储到 TimestampedKeyValueStore 中，但我的消息有一个标头（不仅是类型 id），在它们被存储并沿线发送之后，

Kafka Streams API 中的节点断开连接错误

我已经使用 Kafka Streams API 创建了 Kafka Consumer 代码，并且我能够成功从 Kafka Topic 获取记录并能够处理这些记录。 我在应用程序日志中看到以下错误...

我怎样才能获得具有exactly_once_v2保证的Kafka Streams应用程序以与AWS MSK一起使用

当我配置了启用processing.guarantee=exactly_once_v2 的Spring Boot Kafka Streams 应用程序时，我在运行它时遇到问题。当我启动我的应用程序时，它最终崩溃了（我所有的...

Kafka 流重新分区将数据倾斜到单个分区 - 导致性能问题

我有一个用例，其中我必须根据有效负载值中的参数来计算事件数 流中的数据类似于 键 -> 字符串（不同且可以很好地扩展且唯一） 值 ->