我理解重新平衡可以在您的流上的任何时间和任何点发生。当它这样做时，由于没有为给定的偏移量提交最新的偏移量，可能会发生事件的重新处理。Kafka Streams是否允许在重新平衡发生之前完成任何动态处理？我的意思是，您的应用程序正在使用一个记录(在您的流程方法中)，一个重新平衡事件发生了。因此，这意味着当重新平衡完成时，新分区将具有存储中的最后一个值，但仍然可能“重新处理”该主题的消
浏览 35提问于2020-01-14得票数 1
我想知道Kafka streams的行为(我使用的是低级别Java )。我正在通过实现接口Stream processor来实现org.apache.kafka.streams.processor.Processor。org.apache.kafka.streams.processor.Processor有以下方法：空洞关闭()；
假设kafka主题有6个分区现在，这将触发一个重新</
浏览 4提问于2020-02-24得票数 0
我们开始使用Kafka streams，我们的服务是一个非常简单的无状态消费者。
我们对延迟有严格的要求，当消费群体重新平衡时，我们面临着过高的延迟问题。在我们的场景中，重新平衡将相对频繁地发生:滚动更新代码，扩展/缩减服务，容器被集群调度器洗牌，容器死亡，硬件故障。我们还做了一些滚动更新代码的测试，结果更糟，因为我们的部署没有为Kafka服务做好准备，我们触发了很多重新平衡。我们需要在重新平衡</
浏览 3提问于2019-01-16得票数 4
在我的kafka streams应用程序中创建状态存储时，我得到了这个错误Failed to lock the state directory: /tmp/kafka-streams/string-monitorat org.apache.kafka.streams.processor.internals.AbstractTask.<init>(AbstractTask.java:71)
at org.apache.kafka</
浏览 19提问于2016-08-30得票数 4
当kafka streams应用程序从kafka changelog主题恢复StateStore时，它会处于什么状态？StreamThread将处于哪种状态？如果恢复花费的时间太长，是否会触发重新平衡？
浏览 42提问于2020-01-20得票数 0
但第二个消费者在再平衡期间抛出了一个例外，称其未能实现再平衡。这是因为更改日志的状态发生了更改(下面的异常共享)。我假设，当重新平衡发生时，第一个使用者收到了一些更新，因此更新被推送到相应的更改日志中。请帮帮忙。还共享相同的示例代码。我使用的是Kafka 2_11 0.10.2.1，主题有72个分区 Exception in thread "StreamThread-1" org.apache.kafka.streams.errors.
浏览 2提问于2017-08-30得票数 1
回答已采纳
我知道可以为kafka streams配置复制因子这些内部主题，我们的应用程序使用复制因子3作为常规应用程序主题，但到目前为止，我还没有为changelog/repartition主题配置复制因子，而我的假设是，如果一个broker死亡(或由于某些原因领导者改变)，kafka stream应用程序将自动重新平衡到新的领导者。现在，如果内部主题不使用复制因子，我不太确定正在运行的Kafka Stream应用程序是否可以重新平衡到新的领导者。 对
浏览 6提问于2019-03-04得票数 3
我们使用的是kafka-streams 2.3.1，我刚刚注意到，如果broker宕机，streams应用程序似乎满足于尝试永远保持连接。new KafkaStreams(createTopology(), properties()).start() o.apache.kafka.clients.NetworkClient - [AdminClient在这种情况下，流状态正在重新平衡，因此没有好的方法来确定连接是否刚刚断开。 有没有办法为broker连接尝试设置超时或重试次数？
浏览 41提问于2019-12-19得票数 2
我有一个基本的Kafka Streams应用程序，它从in_topic读取数据，执行滚动聚合，并执行连接以发布到out_topic。它已经正常运行了几个星期，但是今天早上崩溃了，不会再启动了。0b437c22c804] State transition from REBALANCING to ERROR
2019-01-21 17:47:32,860 localhost org.apache.kafka.streams.KafkaStreamsat org.apache.kafka.streams
浏览 5提问于2019-01-22得票数 3
2回答
每当我把任何卡夫卡经纪人打倒，它就进入再平衡，它需要大约。30分钟，有时甚至更长的时间再平衡。at org.apache.kafka.streams.processor.internals.StreamTask.commitOffsets(StreamTask.java:307)at org.apache.kafka.streams.pr
浏览 3提问于2017-09-09得票数 6
回答已采纳
我使用的是无状态处理器，使用Kafka streams 1.0和kafka broker 1.0.1max.poll.interval.ms=Integer.MAX_VALUE //使它像我正在做的那样大，它可能需要10分钟的时间来处理1 kafkamax.poll.records=1
尽管这种
浏览 1提问于2018-04-19得票数 1
回答已采纳
当在只有一个Kafka代理的测试设置中启动我们的Kafka Streams应用程序时，我们看到以下错误，大约15次运行中就有1次：
org.apache.kafka.streams.errors.StreamsException我们在调用org.apache.kafka.streams.KafkaStreams#start之前先执行org.apache.kafka.streams.KafkaStreams#cleanUp。每次测试运行
浏览 1提问于2018-10-23得票数 2
我们意识到，这是消费者优雅地退出和重新平衡的时刻。因此，我们有一些问题：-我们预期当消费者优雅地停止时，会立即触发再平衡，这应该是预期的行为吗？-我们怎样才能减少消费者体面退出和触发再平衡之间的时间？有什么取舍？更多不必要的再平衡？关于更多的上下文，我们的Kafka</e
浏览 2提问于2019-01-28得票数 2
在使用Spark Structured Streaming和Kinesis Streams时，我在重新处理流中积累的数据时遇到了读取不平衡的情况(而不是从最新数据读取)。最近，一位团队成员建议改用Kafka。我对Apache Kafka解决这个问题持怀疑态度，因为AFAIK修复我上面描述的读取不平衡的唯一方法是在消费者级别引入某种协调。这就是Apache Flink的kinesis连接器在重新处理Kinesis streams ()时提供对齐的方式
我
浏览 1提问于2021-07-04得票数 1
为什么在启动Kafka流应用程序时，我会断断续续地点击下面的警告消息？通常情况下，当我重新部署时，它就能工作了。Will try re-join the group with current subscription
在我重新启动应用程序之后，上面的警告就消失了&我得到了一个不同的警告，但至少这个应用可以工作！
浏览 1提问于2020-03-25得票数 7
回答已采纳
，因为组已经重新平衡并将分区分配给另一个成员。:49) at org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:268) at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs:259) at org.apache.kafka.streams</e
浏览 4提问于2017-08-09得票数 6
偶尔，我们的应用程序的一个实例会被合法地杀死，这会导致几轮重新平衡，直到旧节点被替换为止。 at org.apache.kafka.streams.processor.internals.StoreChangelogReader.restore(TaskManager.java:328)
at org.apache.kafka.streams.
浏览 4提问于2019-12-04得票数 1
1回答
对于Kafka streams，我面临着一个非常奇怪的问题，当重平衡发生时，我的kafka streams应用程序一直被以下错误反复出现在日志中：
org.apache.kafka.streams.errors.LockExceptionkafka-streams-2.8.1.jar:?]:433) ~[kafka-streams-2.8.1.jar:?]:849) ~[kafka
浏览 5提问于2022-10-31得票数 1
我正在实现一种机制，通过在本地存储或请求远程Kafka实例来提供一些元数据信息。我正在使用Scala和kafka-streams-scala库的2.4.1版本，我正在运行Kafka集群，它创建了一个带有两个分区的测试主题。流实例，我看到了重新平衡和分区重新分配的过程，并且我很好地理解这两个应用程序之间应该在分区之间共享，例如Kafka流应用程序1应该使用分区0，Kafka流应用程序2应该使用分
浏览 2提问于2020-08-03得票数 0
回答已采纳
我正在访问一个状态存储来查询它，并且不得不用一个try/catch块包装store()语句来重试它，因为有时我会得到这个异常： at org.apache.kafka.streams.state.internals.Stream
浏览 2提问于2018-08-25得票数 3
回答已采纳
云服务器
ICP备案
对象存储
云点播
实时音视频
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号