apache-kafka-streams 相关问题

账户借记中的不一致状态 - 使用Kafka交易的信用申请

使用案例：账户A向账户B发送500美元，我们正在使用一个主题：“账户”有多个分区来记录这些事件生产者 - > 1.交易开始2.Account A BalanceA ...

Kafka KStream OutOfOrderSequenceException

我们的应用程序在我们的流代码中间歇性地遇到OutOfOrderSequenceException。这导致流线程停止。实现很简单，2个KStream加入并输出到另一个主题。 ...

为什么窗口现在适用于Kafka Streams？

我在我的eclipse上运行一个简单的Kafka Streams程序，该程序运行成功，但它无法实现窗口概念。我想处理收到的所有邮件...

在马拉松上重试倒退错误而不是重启kafka-streams

我们有高可用性的kafka。我们运行多个kafka流并在多个主题上激发工作。但是，我们发现了一个发送记录的奇怪问题。我们收到的错误就是这样......

当处理器节点的标点符号到达查找状态存储中的数据时，来自哪个分区的数据？

当调用流处理器标点时，我通过调用KeyValueStateStore.all（）从statestore中提取所有数据，我是否从分配给该kafka-streams的所有分区接收所有数据...

Kafka流 - 每个来源的多个主题作为相同的来源或一个主题？

构建Kafka Streams拓扑时，可以通过两种不同的方式对多个主题进行读取：使用相同的源节点读取所有主题。 topologyBuilder.addSource（“sourceName”，...，“...

如何使用单个分区的记录将时间T的窗口分组到单个记录，每个记录具有相同的密钥

假设我有一个事件流。 R1 - {“abc”：“value 1”} R2 - {“abc”：“value 2”} R3 - {“abc”：“value 3”} R4 - {“abc”：“value 4”} in a单个分区。我想要一个来自...的派生事件流

Kstreams如何限制创建中间主题

运营团队是否有办法限制应用程序团队在kafka集群上创建kafka流中间主题？

使用kafka流的Topic1到Topic2

我是kafka流的新手，我想阅读一个主题，并使用kafka streams api在一个新主题中写下它的一部分。我的关键是字符串，价值是Avro有没有我可以的文档/示例...

汇合Kafka Streams - 找不到类io.confluent.connect.avro.ConnectDefault

我正在使用带有查询模式的jdbc源连接器，似乎没有指定的表名，在schema-registry中为记录键和记录值注册的模式为空...

Kafka Stream的交互式调节

我想为我的kafka流创建交互条件。我只想知道是否有可能。示例用例是这样的：我有用户点击事件，流过我的kafka主题。用户定义......

Kafka Streams 2.1.1类转换，同时刷新定时聚合存储

我正在尝试使用kafka流来执行窗口化聚合，并仅在关闭某个会话窗口后才发出结果。为了达到这个目的，我使用了抑制功能。问题是 ...

Kafka Streams - init（）中的转发记录

在自定义Transformer中，我们尝试在init（..）方法中创建和转发一些记录。根据Kafka Streams JavaDoc，这应该是可能的。但是，这不起作用，我们......

Kafka Streams：跨拓扑共享globalStateStore

我有一个Spring启动应用程序，它使用Processor API生成拓扑，并将addGlobalStateStore生成到同一拓扑。我想创建另一个拓扑（因此另一个KafkaStreams）...

为什么kafka stream state dir在/ tmp / kafka-streams中？

我不确定它是否已经回答了。由于我没有得到适当的解释，请在此处发布我的问题。为什么kafka stream state.dir存储在/ tmp / kafka-streams下？我知道我可以改变路径......

Kafka kstream-kstream与滑动窗口内存的使用一起增长，直到OOM

我在使用kstream连接时遇到问题。我所做的是从一个主题我将3种不同类型的消息分离到新流。然后做一个内部连接与两个流创建另一个...

Kafka Streams删除已使用的重新分区记录，以减少磁盘使用量

我们有一个kafka实例，大约有50M记录，每天输入大约10万，所以在kafka-world中没有什么疯狂的。当我们想用我们更复杂的流应用程序之一重新处理这些记录时（...

Kafka Streams：如何在应用再处理之前获得偏移限制以及如何停止它

我正在寻找一种在KafkaStreams中执行重新处理工具的方法，它允许从主题开始重新处理数据（应用一些过滤器并将这些事件的更新版本写入...

KTable与GlobalKTable之间有什么区别，左边是Join（）vs外连接（）？

在Kafka Stream库中，我想知道KTable和GlobalKTable之间的区别。同样在KStream类中，有两个方法leftJoin（）和outerJoin（）。这两者有什么区别......

与Kafka Streams一起加入不同的主题？

在我目前的项目中，我们使用Kafka，Kafka Connect，Elasticsearch创建了一个数据管道。数据最终以“信号主题”为主题，而不是KeyValue形式 ...