可以从kstream应用读取我在kafkaProducer应用中设置的Kafka消息头吗？我的KafkaProducer看起来像这样；我已经在消息中设置了头 public class Producer {
private final org.slf4j.Logger log = LoggerFactory.getLogger应用程序是 public class M
浏览 6提问于2020-07-17得票数 1
1回答
我正在使用KStream.to(“outputt异位”)编写输出主题；在apache文档中提到，它将自动创建传递给to()的主题。如何使用来自该主题的信息？我可以使用consumer.subscribe()来处理输出消息和轮询消息吗？
浏览 0提问于2019-05-31得票数 2
我对Kafka和阅读文档有点陌生。卡夫卡办公网站有一个。其中应用程序被绑定到一个主题，并且消息一到达就被处理。结果会回发到主题或数据库。Spring Kafka注解@KafkaListener具有相同的功能。例如，我尝试过。在这里，我们也会监听一个主题，并在发布内容时对其进行处理。
所以我很好奇地想知道1.这2个有什么不同?
浏览 5提问于2018-01-18得票数 1
在Spring网站()中，列出了要使用的绑定选项。我们有阿帕奇卡夫卡和卡夫卡流选项。
，他们之间有什么区别？，为了什么目的，我们应该在这两者之间做出选择？
浏览 3提问于2021-06-18得票数 0
回答已采纳
4回答
场景： 我正在将数据JSON对象数据写入kafka主题，而读取时我只想读取基于消息中存在的值的特定消息集。我使用的是kafka-python库。示例消息： {flow_status: "completed", value: 1, active: yes}
{flow_status:"failure",value 2, active:yes} 在这里，我只想读取flow_Status为已完成的<e
浏览 78提问于2019-02-18得票数 2
回答已采纳
1回答
我正在构建一个kafka流应用程序，其中几个主题被合并成一个。卡夫卡流有可能吗？编辑:流使用KStream::merge合并并进行状态转换： .merge(kstream3)
浏览 5提问于2022-04-14得票数 0
回答已采纳
我有一个KStream应用程序，在这个应用程序中，我加入了一个从"topic1“读取的topic1和从"topic2”读取的GlobalKTable，然后加入了从"topic3“读取的另一个GlobalKTable当我试图同时将消息推送到所有三个主题时，我会得到以下例外-
org.apache.kafka.streams.errors.Invalid
浏览 0提问于2019-05-06得票数 2
我的系统包括一个向该主题发送消息的Java生成器应用程序和一个Kafka流应用程序，该应用程序从该应用程序中读取并执行每条消息的操作。| ----> | KStream |目前，我只开始使用Kafka该消息到达
浏览 0提问于2019-07-19得票数 2
回答已采纳
主题A上的信息是使用Kafka流读取的。基于一些验证，消息被发布到主题B或主题C中。这很容易使用Kafka的方法实现。KStreamBuilder kStreamBuilder = new KStreamBuilder();
KStream<String, String> kStream = kStreamBuilder.stream(Serdes.String(), Serdes.String(), topicA.split(","))
浏览 0提问于2018-07-19得票数 0
2回答
我正在处理一个我已经创建了管道的用例，该管道将数据从mongo发送到elasticsearch。Mongo -> Spring Boot -> Kafka -> Transformer(KStream) -> Kafka ->消费者(发送给弹性搜索) 我必须计算从Mongo到Elastic search我想使用Kafka头，并在下一个Kafka生产者中继续转发它们的值，最后通过从当前时间戳中减去来计算相同
浏览 23提问于2019-04-28得票数 1
我们正在尝试实现下面描述的用例，我们有我们希望克服的实现问题，我们正在尝试通过匹配两个流的消息中存在的关键字(JSON)来在两个Kafka主题之间进行KStream连接。此外，我们应该维护消息序列，因为它是从源文件到达KStream的。问题1:如何在其他主题
浏览 2提问于2019-05-08得票数 2
作者在谈到消费者时说：
read_committed:除了读取不属于事务的消息之外，还可以在提交事务之后读取这些消息也就是说，如果只想从主题读取提交写入，则普通使用者应用程序必须指定read_committed。然而，关于Kafka流：
要使流应用程序精确地使用一次语义，只需设置此配置“processing.g
浏览 0提问于2019-06-26得票数 1
回答已采纳
我有一个场景，我通过读取Kafka主题中的数据来创建一个kafka KStream。KStream记录是key=null，value= {a json object}，例如 null: { "ID":"1", "name":"XDFER"}
null: { "ID":"1", "name":"TRAFD"} 真实姓名在GlobalKTable<em
浏览 19提问于2019-03-11得票数 0
回答已采纳
我有一个使用Kstream的处理器的spring云流应用程序。此处理器从异常主题读取消息，并基于每个消息中可用的标志创建两个kstream分支，用于重试主题和拒绝主题。现在需要做的是，在retry主题中出现的任何消息都必须等待一段特定的时间才能将其推回处理主题。谁能帮助我在春季云流中的kafka流应用程序中进行最好的设计或解决方案？
浏览 2提问于2021-06-21得票数 0
我的应用程序从IBM MQ获取消息，并根据内容将其路由到不同的Kafka主题。Kafka MQ源连接器可以将事件从MQ带到1个kafka主题，我们可以在Kafka MQ连接器内进行基于消息的路由吗？
或者我们是否必须编写一个KStream应用程序来根据内容有效负载进行路由
浏览 0提问于2019-08-08得票数 0
当我运行应用程序并向这两个主题发送消息时，join将按预期的方式工作。.主题(流主题)发送消息时，应用程序将使用sensorValues关闭查看堆栈跟踪，我发现由于SensorMetaData中的模式更改，它在执行联接时未能反序列化SensorMetaData。反序列化方法中的断点显示，它试图从主题"app-KSTREAM-TOTABLE-STATE-STORE-0000000002-changelog".反序列化数据
所以问题是为
浏览 2提问于2020-06-06得票数 0
回答已采纳
2回答
给定:我在Kafka中有两个主题--假设主题A和主题B。-- Kafka从主题A读取一个记录，处理它并生成多个记录(比如recordA和recordB)，这些记录对应于消耗的记录。现在，问题是我如何使用卡夫卡流来实现这个目标。KStream<String, List<Message>> producerStreams[] = recordStream.mapValues(new ValueMapper<Message, List*someFunction*(
浏览 3提问于2017-05-30得票数 6
回答已采纳
1回答
我正在编写一个使用Spring云流Kafka流绑定器的kafka流应用程序。@Bean return(input) -> {
KStream<
浏览 1提问于2020-06-07得票数 0
1回答
我试图使用函数编程(和spring云流)从输入主题转换输入AVRO消息，并在输出主题上发布新消息。这是我的转换函数：public Function<KStream<String, Data>, KStream<String, Double>> evenNumberSquareProcessor() {
return kStream -> kStream.transform(() -> new Cust
浏览 5提问于2021-04-06得票数 1
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
