《stream》专题
-
2个具有相同消费者组id的spark stream作业
我正在尝试对消费者群体进行实验 这是我的代码片段 } 当我同时运行两个spark流媒体作业时，它会出错 线程“main”java中出现异常。lang.IllegalStateException:当前没有分配给组织上的分区venkat4-1。阿帕奇。Kafka。客户。消费者内部。订阅状态。组织上的assignedState（SubscriptionState.java:251）。阿帕奇。Kafka。
-
Spring Cloud Stream功能生产者/消费者
我的应用程序由一个带有POST方法的REST控制器组成，用于提交我必须使用生产者发送到主题的数据。 这是控制器 使用Spring-Cloud-Stream版本 从3.1版开始，和注释被弃用，所以我尝试切换到新的方式来设置生产者，我就是这样工作的 最后在应用程序中。yaml我有这个 现在的问题是，当我启动应用程序时，方法被无限调用（我在主题中看到消息）。然后使用供应商似乎我被迫在供应商内部定义消息数
-
用于添加全局存储的Kafka streams用例
在kafka流中定义拓扑时，可以添加全局状态存储。它需要一个源主题以及。处理器接收记录，并在将其添加到存储区之前对其进行理论上的转换。但是在恢复的情况下，记录直接从源主题（changelog）插入到全局状态存储中，跳过在处理器中完成的最终转换。 StreamsBuilder#AddGlobalStore（StoreBuilder StoreBuilder,String topic,Consumen
-
如何暂停（打开/关闭）流处理W/Spring Cloud Streams&Kafka Streams绑定器？
到目前为止我试过什么？-使用/Actuator/Bindingsendpoint启动/停止kafka-streams绑定。这似乎不适用于Kafka-Streams活页夹，只适用于Kafka活页夹：（。 任何帮助都将不胜感激！谢了！
-
使用kotlin streams从列表中获取随机项
如何以简单简洁的方式从列表中获取随机项<如果我想从这个列表中得到一个偶数随机数，那么。 注意： 我知道java中有一些类似的答案可以解决这个问题，但我认为我们可以在kotlin中有一个更简洁的方法。
-
仅使用spring cloud stream kafka streams绑定器自动创建生产者主题
是否可以在application.yml中只为应用程序生成的主题启用，而不是自动创建应用程序使用的主题？我希望我的制片人只负责创造话题。我正在使用spring-cloud-streams和kafka-streams绑定器。 从更广泛的意义上说，这是创建Kafka主题的正确方法吗？
-
用Spring cloud stream Kafka绑定器在批处理功能中发布多条消息
-
如何通过属性设置Spring Cloud Stream Kafka的偏移提交间隔
我想通过属性为Spring Cloud Stream ；在Kafka consumer上设置一个自动偏移提交间隔。 正如我从度量中看到的，默认情况下，Spring Cloud Stream ；Kafka对每个消耗的消息提交偏移量。对于高负载的主题（例如，如果流量为每秒10K消息），它会变得非常明显，并增加Kafka broker的负载。 我们以以下方式声明消费者： 我尝试了几个选择，但都没有帮助我
-
升级到2.1后使用spring-cloud stream和kafka的项目打开的文件太多
在使用Multibinder，kafka(2个代理）和rabbit(1个代理）的项目上升级到spring cloud Stream2.1后，我们面临着打开文件太多的问题。 打开的文件数量一直增长到操作系统(RedHat7.3)定义的限制。 在第一次尝试中，我想用kafka-clients 1.0.2来尝试spring cloud Stream2.1，根据文档，这是可能的，但我面临一个问题。下面是我
-
Spring Cloud Stream2.0是否兼容Kafka Broker0.10.2？
那么Spring-Cloud-Stream-Binder-Kafka2.0和0.10.2代理之间是否存在兼容性问题呢？ 我计划从spring-cloud-stream 1.2+Kafka 0.10.2升级到spring-cloud-stream 2.0+Kafka 1.0，并试图了解是否可以一次性完成（客户机->1.0，然后代理->1.0)，或者spring-cloud-stream支持的无停机升
-
Spring Cloud Stream Kafka活页夹：“试图从状态IN_TRANSACTION转换到状态IN_TRANSACTION无效”
我在生产者中安装了Apache Kafka“Kafka2.11-1.0.0”并定义了“TransactionIDPrefix”，我知道这是在Spring Kafka中启用事务所需要做的唯一一件事，但是当我这样做并在同一个应用程序中运行简单的源和接收器绑定时，我会看到消费者收到并打印了一些消息，而有些消息会出现错误。 例如，收到的消息#6： 但是消息#7有一个错误“试图从状态IN_TRANSACTI
-
带有spring cloud stream的Kafka客户端id属性
有了Spring Cloud Stream，有什么方法可以设置Kafka client id属性吗？当然我用的是Kafka活页夹。我的几个设置此设置的服务将被复制，那么有什么方法可以使IDs在实例之间唯一呢？（查看Spring Kafka Consumer Client-Id配置，似乎IDs需要是唯一的。） 假设我可以设置该属性，那么在运行时是否需要查询ID？
-
如何使用stream API按指定的顺序对映射进行排序？[关闭]
映射需要在降序时按值排序，然后在升序时按键排序。方法将与作为参数添加到方法中的两个映射相关的值之和添加到方法中，并将其作为自己的值存储。 输出应该是：伯格达到了产能！用户数：3克拉克-8马克-9凯文-1 然而我得到的是：伯格达到了容量！用户数：3马克-9克拉克-8凯文-1 我做错了什么？
-
用Kafka Streams绑定器和函数式处理器处理Spring Cloud Streams处理异常实例
但是：我想集成异常处理，处理处理器中发生的异常，并将这些不可处理的消息发送到DLQ。我已经为反序列化错误设置了DLQ，但除了SobyChacko对类似问题的回答之外，我没有找到关于实现这一点的好建议。但这只是一个片段！有人有更详细的例子吗？我之所以问这个问题，是因为Spring Cloud Stream关于分支的文档看起来完全不同。
-
Spring Cloud Stream消息正文中发现的嵌入式头
我使用Spring Cloud Stream 1.3.2.Release向Kafka发布字符串消息。当我使用命令行Kafka consumer或Spring Kafka使用消息时，总是在消息正文中追加一个contentType头。 问题： 有什么方法可以去掉嵌入的头吗？ --