学习apache-kafka-streams－开发者之家

问题描述我正在尝试将StreamsUncaughtExceptionHandler添加到我的Kafka流处理器中。该处理器是用Kafka函数编写的。我查看了suggestionprovidedbyArtemBilan以将StreamsUncaughtExceptionHandle

问题描述我正在尝试KafkaStreams的字数统计问题。我正在使用带有Scala版本2.11.12和SBT版本1.1.4的Kafka1.1.0。我收到以下错误：Exceptioninthread"wordcount-application-d81ee069-9307-46f1-

问题描述默认情况下，.windowedBy(SessionWindows.with(Duration.ofSeconds(60))为每个传入记录返回一条记录。结合使用.count()和.filter()可以轻松检索第一条记录。使用.suppress(Suppressed.unti

问题描述我有一个Kafka流--比如博客和Kafka表--比如那些博客相关的评论。Kafka流中的key可以映射到Kafka表中的多个值，即一个博客可以有多条评论。我想连接这两个对象，并创建一个带有注释ID数组的新对象。但是当我连接时，流只包含最后一个注释id。有没有任何文档或示

问题描述我有一个KafkaStreams应用程序，它从几个主题中获取数据，并连接数据并将其放入另一个主题中。卡夫卡配置：5kafkabrokersKafkaTopics-15partitionsand3replicationfactor.注意：我正在运行KafkaBroker的同

问题描述如@https://kafka.apache.org/21/documentation/streams/developer-guide/dsl-api.html#window-final-results所述，我有一个要求，即等待窗口关闭，以便通过在窗口持续时间内对其进行缓

发布：2022-10-16 标签：apache-kafka-streams

问题描述我正在为我的SpringKafkaStreams应用程序的定制而苦苦挣扎。我一直在尝试在我的KStreams中配置未捕获(运行时异常)处理。参照文档https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/3

问题描述我使用的是SpringCloudStream和KafkaStream。假设我有一个处理器，它是一个将字符串的KStream转换为CityProgrammes的KStream的函数。它调用一个API来按名称查找City，并调用另一个转换来查找该城市附近的任何事件。现在的问题

问题描述我们有一个Streams应用程序，它使用源主题中的消息，执行一些处理并将结果转发到目标主题。消息的结构由某些Avro架构控制。当开始使用消息时，如果架构尚未缓存，应用程序将尝试从架构注册表中检索它。如果由于任何原因架构注册表不可用(例如网络故障)，则当前正在处理的消息将丢

问题描述我们可以在不聚合的情况下应用Kafka窗口操作吗？我需要过去10分钟内的所有数据(而不是计数)？我所看到的是，大多数示例使用了窗口滑动的聚合。推荐答案您可以使用KafkaConsumer#offsetsForTimes方法获取分区到偏移量映射中10分钟前的偏移量。使用该信