1回答
目前使用Kafka Streams在客户端系统中聚合事件。当使用假事件运行我们的原型时，一切都运行得很完美。然而，在使用实际数据时，我们注意到在聚合过程中，Streams会自动创建内部主题。虽然在理论上这是很好的，我们的客户有必要的，超级严密的安全性，不愿意授予我的开发团队主题创建特权。这意味着我们不能按原样运行Streams程序。不过，我们可以为自己创建主题，并使用这些主题来代替创建自己的Kafka主题</
浏览 21提问于2021-06-08得票数 0
1回答
在我的Springboot应用程序中，我使用了kafka-streams:2.7.0。我必须加入两个KStreams并更改内部主题名。StreamJoined.with(Serdes.String(), docSerde, docSerde).withStoreName("test123"))kafka
浏览 2提问于2022-04-04得票数 0
我们通过集成使用Kafka流。我将复制因子配置为在所有内部Kafka流主题中使用，方法是设置它适用于大多数由Kafka流在内部使用的重新分区/变更主题。对于这些主题，我仍然可以看到复制因子设置为默认1，也不可能使用Material
浏览 5提问于2020-10-14得票数 1
回答已采纳
然后，我按照链接添加了以下配置：然后应用程序可以正常工作，但是当创建了状态存储内部主题时，但是当Kafka重新启动并且状态存储主题丢失/删除时，它可以正常工作，然后Kafka流处理器需要在启动应用程序时自动创建内部状态存储主题，此时它会抛出异常，即：
"Aorg.apache.kafka.streams.errors.StreamsException
浏览 3提问于2020-06-04得票数 1
我使用kafka流约函数，它创建了一些状态存储、更改日志、kafka内部主题(比如app-KSTREAM-REDUCE-STATE-STORE-0000000002-changelog )。因此，我在kafka代码中设置了以下内容：props.put(StreamsConfig.TOPIC_PREFIX + TopicConfig.RETENTION_BYTES_CONFIGStreamsConfig.TOPIC_PREFIX +
浏览 0提问于2018-09-02得票数 6
回答已采纳
根据文档()，内部主题遵循命名约定<application.id>-<operatorName>-<suffix>。不幸的是，我们的安全需求不允许我们用我们的应用程序创建主题，并且需要提前设置。我正在尝试确定这些主题名称是否会一致。
浏览 2提问于2018-10-04得票数 4
当在只有一个Kafka代理的测试设置中启动我们的Kafka Streams应用程序时，我们看到以下错误，大约15次运行中就有1次：看起来主题是用预期的分区数(32)创建的。稍后，在相同的日志中，看起来有一个请求再次创建该主题。我们从来没有在非测试中看到过这种情况，我们使用6个Kafka代理
浏览 1提问于2018-10-23得票数 2
Kafka stream自动设置内部主题的retention.ms和cleanup.policy，例如。物化的KTables。但是，我注意到WindowStoreBuilder创建的日志主题上没有设置retention.ms。 直观地说，日志主题也是内部主题，可以/应该是自动配置的。
浏览 14提问于2020-07-09得票数 2
我开始研究如何重新划分压缩的主题，方法是使用一个groupBy()，后面跟着一个()，但是当它开始读取创建的重新分区主题时，它开始抛出一个带有消息的：
在org.apach
浏览 1提问于2019-08-28得票数 0
回答已采纳
我试图使用kafka streams库来使用kafka的一次功能。我只将proessing.guarantee配置为exactly_once。此外，还需要将事务状态存储在内部主题(__transaction_state)中。
我的问题是，如何自定义主题的名称？如果kafka集群被多个消费者共享，每个客户是否需要不同的交易管理主题？
浏览 2提问于2018-08-16得票数 0
1回答
有两种方法，我可以想到：
流处理:使用Kafka Streams API来卷起数据，并在Cassandra中摄取汇总的数据。
浏览 0提问于2019-03-25得票数 1
我知道可以为kafka streams配置复制因子这些内部主题，我们的应用程序使用复制因子3作为常规应用程序主题，但到目前为止，我还没有为changelog/repartition主题配置复制因子，而我的假设是，如果一个broker死亡(或由于某些原因领导者改变)，kafka stream应用程序将自动重新平衡到新的领导者。现在，如果内部主题不使用复制因子，我不太确定正在运行的Kafka Stream应用程
浏览 6提问于2019-03-04得票数 3
我有一个卡夫卡主题，名为input with multiple partitions。假设一条消息是这样的： "key": 123456, ... 记录按“键”进行分区(因此，相同的键将始终由相同的Kafka使用者处理untilWindowCloses(Suppressed.BufferConfig.unbounded())) .to("output");
有了groupBy，
浏览 0提问于2018-12-30得票数 3
可以为卡夫卡流提供像"retention.ms“、"cleanup.policy”这样的主题配置，比如*-changelog主题来删除无用的日志。但是，对于内部主题，如*-重新分区主题，不可能提供主题配置值，尽管重新分区主题的默认"retention.ms“是"-1”，这意味着无限保留。如何删除或管理重新分区主题？否则，重新分区主题的大小将太大，可能会出现磁盘故障问题。
如何管理重新分区
浏览 5提问于2021-01-30得票数 3
回答已采纳
1回答
所有输入、中间和输出主题都是基于avro的主题。如何处理拓扑的更改(添加新步骤、新的输入-输入-输出主题)，了解kafka-streams维护内部状态和changelogs，符合以下业务要求：
没有停机时间答案是介于使用kafka-streams-application-reset还是创建全新的消费者组(应用程序id)之间吗？
浏览 1提问于2018-12-12得票数 2
回答已采纳
应用程序/环境详细信息：Caused by: org.apa
浏览 1提问于2021-09-17得票数 0
假设我们正在KStream和KTable之间进行内部连接，如下所示： JsonSerde当我运行应用程序并向这两个主题发送消息时，join将按预期的方式工作。.主题(流主题)发送消息时，应用程序将使用sensorValues关闭查看堆栈跟踪，我发现由于SensorMetaData中的模式更改，它在执行联接时未能反序列化SensorMetaData。反序列化方法中的断点显示，它试图从主题&qu
浏览 2提问于2020-06-06得票数 0
回答已采纳
当您运行Kafka streams应用程序重置工具以将应用程序重置为特定的时间戳(例如T-n)时，状态存储会发生什么？文档内容为：“内部主题:删除内部主题(这会自动删除任何已提交的偏移量)”(内部主题在执行状态存储的changelog主题时由Kafka Streams应用程序在内部使用)
这是否意味着我丢失了状态存储/
浏览 13提问于2020-02-25得票数 1
回答已采纳
当源主题分区计数= 1时工作正常。如果我将分区增加到任何大于1的值，我会看到下面的错误。既适用于低级，也适用于DSL API。有什么建议吗？可能会遗漏什么？org.apache.kafka.streams.errors.StreamsException: stream-thread [StreamThread-1] Failed to rebalanceat org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread
浏览 3提问于2017-02-19得票数 6
回答已采纳
kafka流如何保证由kafka broker分配给它的分区与分配给其他主题的分区匹配？好像需要协调一下吗？而且，卡夫卡流总是从一开始就阅读压缩的主题，还是从最新的偏移量中读取？一旦它从压缩的主题中读取消息，它是否提交偏移量？
浏览 2提问于2020-07-31得票数 0
回答已采纳
云服务器
ICP备案
云直播
即时通信 IM
实时音视频
腾讯云开发者
