我需要使用多个转换器在kafka上解析复杂的消息。每个转换器解析消息的一部分，并通过填充消息上的某

<strong>问题：</strong>：我有一个用例，想在每个窗口的Kafka Stream中获取主题名称，分区ID等。 <stron

我被困在一个用例中。我需要计算娱乐花费的百分比。 在我的信息流中，我获得了包含行业代码和

因此，我构建了一个流应用程序，将来自源主题的数据存储到带有时间戳的KeyValue状态存储中。在启动流

我使用水槽将数据发布到2个kafka主题，如何在单独的数据框中读取它们？ 我使用了2个主题，akr和ak

我已经在我的kafka流实例上附加了<strong> org.apache.kafka.streams.processor.StateRestoreListener </strong>接口的实现。

这是一个假设的场景，但与我使用Kafka Streams试图实现的场景类似。 我有流数据，销售，我想通过

我有一个执行某些状态处理的微服务。该应用程序从输入主题构造一个KStream，进行一些有状态的处理，

我试图查看GlobalKtable在kafka中的工作方式，为此，我试图编写示例代码。 我已经创建了globalKtable，但我

我有一个流，它使消息映射到两个不同的map（）调用，并进一步被过滤并写入两个不同的主题。 <pre><

有人可以给我一个示例实例，其中我们应该从kafka中读取数据作为Ktable，又一个实例中我需要从kafka中读

我有一个基于Kafka流的应用程序（使用低级API），简而言之，它执行以下操作： 流处理（具有<code>p

当kafka-streams应用程序正在运行并且Kafka突然关闭时，该应用程序进入“等待”模式，在其上发送警告日

在kafka Streams（版本：2.3.1）上，我们面临的是提交偏移量的问题： <blockquote> org.apache.kafka.common.erro

我的流拓扑如下： <pre><code>Topology topology = new Topology(); //WS connection processor topology.addSource

我有一个Kafka主题，可以通过设置cleanup.policy = compact启用压缩。 我的segment.bytes属性设置为稍大的值（100M

我有一个Spring boot Kafka Stream应用程序，该应用程序处理所有传入事件并将其存储在Kafka Streams内部提供的

我正在使用<strong> Spring Cloud Stream Kafka Streams </strong>编写Java应用程序。这是我正在使用的功能方法片段：

就像我在标题中说的那样，当生产者停止发送经期信息时，我想接收最后的windowBy消息。目前，我正在手

我显然是kafka / kafka流的初学者。我只需要阅读来自几个主题的给定消息，并给出其ID。虽然我们的实际