kafka是否在重新平衡已传输分区时应用压缩
Kafka在重新平衡已传输分区时不会应用压缩。重新平衡是指当Kafka集群中有新的Broker加入或者有Broker离开时，Kafka会重新分配分区以实现负载均衡。在重新平衡期间，Kafka会将分区从一个Broker迁移到另一个Broker，以确保每个Broker上的负载均衡。
在重新平衡过程中，Kafka并不会应用压缩。压缩是一种数据压缩技术，可以减小数据的存储空间和传输带宽。在Kafka中，压缩通常是在生产者端进行的，生产者可以选择将消息进行压缩后再发送到Kafka集群。而在重新平衡过程中，Kafka只是简单地将分区从一个Broker迁移到另一个Broker，并不会对消息进行压缩操作。
需要注意的是，Kafka本身并不提供压缩功能，而是通过使用压缩算法的方式来实现消息的压缩。常见的压缩算法有Gzip、Snappy和LZ4等。这些压缩算法可以在生产者端进行配置，以实现对消息的压缩。在消费者端，Kafka会自动解压缩消息，使其恢复为原始的消息格式。
总结起来，Kafka在重新平衡已传输分区时不会应用压缩。压缩是一种在生产者端进行的操作，而重新平衡只是简单地将分区从一个Broker迁移到另一个Broker，不涉及消息的压缩操作。
相关·内容
1)采用底层处理器API。当重新平衡发生时，处理器对象是否被重新初始化并再次调用init方法？
( 2)对于整个Kafka流，是否是特定于重新平衡分区(单个应用程序具有多个线程占用不同的分区)，还是因为它是一个流状态？
浏览 5提问于2020-03-03得票数 1
2回答
当添加或关闭新的consumer/brorker时，Kafka会触发重新平衡操作。Kafka是否正在重新平衡阻塞操作。在重新平衡操作进行期间，Kafka消费者是否被阻止？
浏览 1提问于2014-11-28得票数 36
回答已采纳
我搞不懂分区分配在多大程度上是客户端关心的partition.assignment.strategy，以及哪些部分是由Kafka处理的。
例如，假设我有一个包含100个分区的kafka主题。
如果我制作一个应用程序，运行5个线程的消费者，partition.assignment.strategy为RangeAssignor，那么我应该得到5个消费者，每个消费者消费25个分区。
现在，如果我通过部署该应用程序4次，并使用相同的消费者组来扩展它。kafka是否会先将25个分区划分为其一侧的每个应用程序，然后使用PartitionStrategy的应用程序才会进一步细分这25个分区？
这将整齐地产生
浏览 2提问于2018-12-19得票数 2
2回答
Kafka Broker有3个分区。
卡夫卡消费者实例的计数是3。
突然，一个消费者的例子死了。
我知道，如果一个Kafka消费者实例死亡，Kafka代理正在进行再平衡，而另一个消费者实例将被分配到该分区。
我想知道这样的假设是否正确:另一个实例使用它最初使用的所有分区，然后分配和使用死分区。
(我是否必须在客户端代码中实现ConsumerRebalanceListener？)
如果是这样的话，在使用消息时会有什么延迟吗？
谢谢。
浏览 3提问于2017-05-17得票数 1
我有一个应用程序下载特定的网页内容，从一个从1 Kafka产生的URL流-生产者。我已经创建了一个有5个分区的主题，有5个kafka-consumers。然而，网页下载的超时时间是60秒。当其中一个url正在下载时，服务器会假定该消息已丢失，并将数据重新发送给不同的消费者。
我试过了我提到的所有东西
和
但我每次都收到不同的错误。
是否可以在kafka中将特定的消费者与分区绑定？我在我的应用程序中使用kafka-python
浏览 4提问于2017-08-29得票数 4
回答已采纳
我用java构建了一个kafka应用程序，它具有：
是一家生产卡夫卡唱片的生产商--卡夫卡是一种卡夫卡流，使用这些记录，对其值应用一些(时间窗口和状态存储)操作，并将它们发送回kafka ，消费者消费这些转换后的值，并将它们写入数据库。
我是测量卡夫卡时间戳的生产者(它被卡夫卡流消费)和消费者记录(它被消费者消费)之间的时间差。所以，基本上，当制作人的唱片被创建，当这张唱片被流转换，并被送回卡夫卡时。最后，我取了数据库中每个时差的平均值。
由于任何原因，当我向主题添加更多流节点和更多分区时，这种时间差就会增大。实际上，我预计时差会减小。现在，我想知道我是否做错了什么，或者通过增加节点数量，数据
浏览 3提问于2020-12-03得票数 2
我们正在运行一个3代理Kafka 0.10.0.1集群。我们有一个java应用程序，它可以从不同的主题中产生许多消费者线程。对于每个主题，我们指定了不同的消费群体。
很多时候，我发现只要重启这个应用程序，一个或多个CG接收分区分配的时间就会超过5分钟。在此之前，该主题的消费者不会消费任何东西。如果我转到Kafka broker并运行consumer-groups.sh并描述那个特定的CG，我会看到它正在重新平衡。在server.log中，我看到这样的代码行
准备稳定组otp-sms-consumer稳定组otp-sms-consumer
这两个日志之间通常会有大约5分钟或更长的间隔。在消费者端
浏览 2提问于2018-07-13得票数 10
让我们假设我有一个有状态的Kafka应用程序，它使用来自主题的3个分区的数据。目前，我有两个以上应用程序的实例正在运行。让我们这样说：instance1有分区part1和part2，instance2有part3。
所以现在我想要添加新的实例来完全利用并行化。
据我理解，一旦启动新实例，就会发生重新平衡:一个分区( part1或part2 )和相应的本地状态存储将从现有实例迁移到新添加的实例。在这个例子中，让我们假设part1在instance3上迁移。
同时，我意识到新实例instance3将不会开始处理新的数据，直到它从changelog主题中恢复本地状态存储，这可能需要很长时间。
从启动
浏览 4提问于2021-11-03得票数 0
回答已采纳
Kafka重新平衡算法可以跨主题工作吗？
假设我有5个主题，每个主题有10个分区，同一消费者组中有20个消费者应用程序实例，每个实例都订阅了这5个主题。
Kafka会尝试在20个实例中均衡50个分区吗？
或者它是否只在一个主题内平衡，因此10个第一实例可能(或可能)接收所有50个分区，而其他10个实例可能保持空闲？
我知道在过去的日子里，Kafka不会平衡不同的主题，但是现在的版本呢？
浏览 2提问于2020-09-13得票数 2
我有点搞不懂这在概念上是如何运作的。
kafka流如何保证由kafka broker分配给它的分区与分配给其他主题的分区匹配？好像需要协调一下吗？而且，卡夫卡流总是从一开始就阅读压缩的主题，还是从最新的偏移量中读取？一旦它从压缩的主题中读取消息，它是否提交偏移量？
浏览 2提问于2020-07-31得票数 0
回答已采纳
我们使用的是spring，kafka(spring kafka 2.4.5，spring cloud stream 3.0.1)，openshift。我们有以下配置。多个代理/主题，每个主题有8个分区，复制因子为3，多个spring boot消费者。
作为弹性测试的一部分，当我们关闭其中一个代理时，我们会得到下面的异常，甚至在我们启动服务器之后，仍然会得到相同的错误。
org.apache.kafka.common.errors.NotLeaderForPartitionException: This server is not the leader for that topic-partit
浏览 287提问于2020-05-20得票数 1
我想知道Kafka streams的行为(我使用的是低级别Java )。我正在通过实现接口Stream processor来实现org.apache.kafka.streams.processor.Processor。
org.apache.kafka.streams.processor.Processor有以下方法：
无效init(ProcessorContext上下文)；
空洞关闭()；
假设kafka主题有6个分区，在我的流应用程序中，它有num.stream.threads=1，并在不同的机器上运行SIX instances of streaming application。因此，这意
浏览 4提问于2020-02-24得票数 0
2回答
我有一个Kafka流应用程序，它从几个主题中获取数据，并将数据加入到另一个主题中。
Kafka配置：
5 kafka brokers
Kafka Topics - 15 partitions and 3 replication factor.
注:我在卡夫卡经纪人运行的同一台机器上运行卡夫卡流应用程序。
每小时消耗/产生数以百万计的记录。每当我把任何卡夫卡经纪人打倒，它就进入再平衡，它需要大约。30分钟，有时甚至更长的时间再平衡。
有人知道如何解决卡夫卡消费者的再平衡问题吗？而且，它多次在重新平衡的同时抛出异常。
这将阻止我们在生产环境中使用此设置。任何帮助都将不胜感激。
Caused by
浏览 3提问于2017-09-09得票数 6
回答已采纳
1回答
我有一个定制的kafka设置，我的应用程序和kafka代理放在一个节点中。
为了确保app实例只使用该节点中的分区(以减少网络开销)，我将一个自定义分区分配给组的所有成员。
然而，如果代理失败，然后重新加入集群，这会引发消费者再平衡吗？类似地，如果我添加一个新代理并触发分区重新分配脚本，这也会触发重新平衡吗？
浏览 1提问于2020-10-20得票数 1
回答已采纳
我们开始使用Kafka streams，我们的服务是一个非常简单的无状态消费者。
我们对延迟有严格的要求，当消费群体重新平衡时，我们面临着过高的延迟问题。在我们的场景中，重新平衡将相对频繁地发生:滚动更新代码，扩展/缩减服务，容器被集群调度器洗牌，容器死亡，硬件故障。
我们完成的第一个测试之一是让一个由4个消费者组成的小型消费者组处理少量的消息(1K/秒)并杀死其中的一个；集群管理器(目前是AWS-ECS，可能很快就会转移到K8S)启动一个新的。因此，不止一次完成了重新平衡。
我们最关键的指标是延迟，我们将其度量为发布者中的消息创建和订阅者中的消息消耗之间的毫秒。我们看到最大延迟从几毫秒增加到
浏览 3提问于2019-01-16得票数 4
我正在做一个卡夫卡消费者项目。最近我们在PROD环境中部署了它。在那里，我们遇到了如下问题：
[main] INFO com.cisco.kafka.consumer.RTRKafkaConsumer - No. of records fetched: 1
[kafka-coordinator-heartbeat-thread | otm-opl-group] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-otm-opl-group-1, gr
浏览 4提问于2020-03-30得票数 4
1回答
我们使用ApacheKafka2.2版本与5个代理。我们每天得到5000万个事件，但我们正在达到高卡夫卡cpu使用率。我们正在使用默认的生产者/消费者/代理设置。
我对演出有一些疑问；
我们有不同的kafka流应用程序，它们进行聚合，或者加入操作来传递丰富的消息。我们所有的kafka流应用程序都由以下设置组成：
exactly-once:truemin-in-sync replicas:3replication因子:3主题分区：~50~100(拓扑不同)
当然，可能会有内部的拓扑主题。我们正在将我们的工人机器扩展到至少5个应用程序中。因此，每个实例在大多数情况下都会为一个分区策略寻找一个线程。
浏览 1提问于2020-08-01得票数 0
我使用Spring kafka 1.0.3来消费kafka消息。2个kafka主题，每个主题有1个分区。在java代码中，有2个@KafkaListener来消费每个主题消息。ConcurrentKafkaListenerContainerFactory的并发设置为1，但消息有时会延迟20+秒。
@Configuration
@EnableKafka
public class KafkaConfig {
@Value("${kafka.brokers}")
private String brokers;
@Bean
public KafkaListenerContainer
浏览 7提问于2016-09-26得票数 0
我是卡夫卡的一个新用户，已经试用了2-3周了。我相信目前我已经很好地理解了Kafka的大部分工作原理，但是在尝试为我自己的Kafka用户安装API之后(这是模糊的，但我遵循的是新的KafkaConsumer的指南，它应该在V0.9中提供，这是在‘主干’回购ATM机上发布的)，如果我有多个具有相同groupID的用户，我就会出现延迟问题。
在此设置中，我的控制台始终记录有关“再平衡触发”的问题。当我向使用者组中添加新的使用者时，是否会发生再平衡，并且是否触发这些用户实例，以确定同一个groupID中的哪个使用者实例将得到哪些分区，或者是完全用于其他事情的再平衡？
我也从上看到了这段话，但我似乎无
浏览 0提问于2015-06-22得票数 80
回答已采纳
我有一个包含3个节点的Kafka集群。我想向集群中再添加2个节点，使其成为5个节点的集群。添加节点后，我将执行完全集群重启。即一次性关闭所有kafka & zookeeper服务，然后启动5个节点的所有服务。
我想知道1.全集群重启是否将topic分区均匀分布在5个节点上？或者需要运行分区重新分配工具Kafka - reassign -artitions.sh? 2. kafka没有内部机制自动重新分配分区，除非运行kafka-reassign-artitions.sh
请澄清。谢谢。
浏览 3提问于2019-07-12得票数 0
这是如何使用名称- test_test创建新的10个主题分区的示例。
kafka-topics.sh --create --zookeeper zookeeper01:2181 --replication-factor 3 --partitions 10 --topic test_test
Created topic "test_test".
[root@kafka01 kafka-data]# \ls -ltr | grep test_test
drwxr-xr-x 2 kafka hadoop 4096 Mar 22 16:53 test_test-8
drwxr
浏览 3提问于2020-03-22得票数 0
回答已采纳
2回答
我正在尝试对消费者群体进行实验。
这是我的代码片段
public final class App {
private static final int INTERVAL = 5000;
public static void main(String[] args) throws Exception {
Map<String, Object> kafkaParams = new HashMap<>();
kafkaParams.put("bootstrap.servers", "xxx:9092");
kafka
浏览 0提问于2018-05-27得票数 2
回答已采纳
对于特定的主题，我们希望每个代理有一个分区。我们的想法是，为了扩大规模，我们继续增加经纪人。但是，当我测试如何添加新分区时，它的工作方式并不像我所期望的那样。例如，我有三个经纪人: 2064029428,2064029280,2064028458。我创造了一个新的主题：
/opt/kafka/bin/kafka-topics.sh --create --zookeeper zk.net:2181/stream --topic test --partitions 2 --replication-factor 1
/opt/kafka/bin/kafka-topics.sh --describe
浏览 0提问于2014-06-26得票数 3
回答已采纳
我们新的kafka集群空间耗尽的速度比预期的要快，所以我采取了一些措施来缓解：
1)我为有问题的主题打开了快速压缩:这是在代理端使用"kafka-topics -config compression.type=snappy“命令完成的，这样就不会中断生产者
2)我将集群中的kafka节点数量增加了一倍
3)我重新平衡了集群。
这些步骤奏效了，我们赢得了一些时间。我知道1只适用于新生成的数据，但我想知道，重新平衡到新节点的分区在到达新代理时是否会被压缩，或者它们是否会保留其原始形式(没有生产者压缩，因此它们都是未压缩的)。旧的节点上有一些其他的东西，这使得从简单的磁盘使用分析中很难推断出
浏览 6提问于2018-12-13得票数 0
1回答
我使用的是Kafka流，具有无状态的简单处理器拓扑。
我有一个有100个分区的主题，有两台各有50个线程的机器，并且运行相同的流应用程序，所以最终我将有一个1-1之间的映射。
主题中的消息已经是键控消息。
我有一个逻辑约束，即一旦一个线程连接到一个分区或更多的分区，它就应该继续处理这些分区(当然，直到重新启动时，它会重新洗牌)。
我从日志中看到线程多次(重新)加入了使用者组。
我的问题是，kafka-streaming是否保证线程能够处理最初挂在应用程序启动上的相同的分区，或者它不时地进行重新设置。
我检查了文档，但没有找到任何细节上讨论这个问题的东西。
下面是我使用的代码：
StreamsB
浏览 0提问于2018-03-19得票数 1
回答已采纳
1回答
最近，我们将Kafka升级为v1.1，并将其升级为V4.0。但在升级过程中，我们遇到了关于国营商店的持续问题。我们的应用程序启动一个流的集合，并在100次尝试之后，在关闭应用程序之前检查状态存储是否已经就绪。但是升级之后，至少有一个流将具有Store is not ready : the state store, <your stream>, may have migrated to another instance，流本身具有RUNNING状态，消息将通过，但是存储的状态仍然显示为未准备就绪。所以我不知道会发生什么。
我不应该检查存储状态吗？
和由于我们的应用程序有很多流
浏览 1提问于2018-06-05得票数 1
我有一个关于Kubernetes中的Kafka的问题，特别是自动缩放...
假设我在Kubernetes的3个Pod中有3个Kafka Broker，有一个带有5个分区(P1，P2，P3，P4，P5)的TopicA，复制因子是3，所有的Broker都有他们的持久卷，我在Kubernetes中配置了自动扩展，所以如果它检测到Kafka Pods中的%80CPU/内存使用率，它将为Kafka Broker启动额外的Pod...
如果我没有完全错的话，Kafka会检测到Zookeeper上的额外实例，并可以移动分区(所以假设P1，P2在Broker1和P3，P4在Broker2，P5在Broker3
浏览 5提问于2019-11-06得票数 2
我正在研究在我的项目中使用Kafka作为消息代理的可能性，它具有强大的消息处理顺序保证。
我了解到，在Kafka中，主题被分成几个分区，其中每个分区都分配了一个使用者(假设只有一个使用者组)。这对维持秩序很有好处。再平衡是将分区重新分配给使用者的过程，当分区数或使用者数量被更改时(失败，新工作人员添加)，就会触发重新平衡。
是否有可能在不改变消费者数量的情况下利用这种在飞行中的再平衡？
假设我们有以下情况：
P1 -- C1
heavytraffic -> P2 -- C2
heavytraffic -> P3 --/
因此，使用者C1只处
浏览 0提问于2020-09-21得票数 0
回答已采纳
2回答
星火流的Kafka分区
、、、
我使用的卡夫卡与火花流(2.2.0)。系统上的负载是动态的，我试图了解如何处理自动缩放。自动缩放有两个方面：
自动缩放下的计算
自动缩放应用程序组件以利用自动缩放下的功能
Infra自动缩放：可以有各种定义良好的触发点来扩展下位。在我的情况下，可能的问题之一是处理到达kafka的消息的延迟或延迟。因此，我可以监视kafka集群，如果消息处理延迟超过某个因素，那么我知道需要投入更多的计算能力。
应用程序自动缩放：在上面的场景中，让我们说，一旦我觉得在卡夫卡中耽搁了很长时间，我就会再给星火集群添加一个nore。一个新的工作人员启动并向主服务器注册，因此星火集群拥有更多可用的马力。利
浏览 2提问于2018-04-17得票数 0
3回答
用于消息处理的多个节点
、、、、
我们在Kubernetes上部署了一个spring引导应用程序来处理消息:它从Kafka主题中读取，然后执行一些映射，最后，它写到Kafka主题。
为了获得更高的性能，我们需要更快地处理消息，因此我们引入了这个春季引导应用程序的多个节点。
但我相信这会导致一个问题，因为：
应按顺序处理消息。
消息包含一个状态。
是否有任何解决方案来保持消息的有序性，并保证一个节点已经处理的消息不会被另一个节点处理，并解决多个节点中的处理所引起的任何其他问题。
请自由讨论所有可能的解决方案，因为我们正在建设一个POC。
使用apache或spring云流对此有帮助吗？
浏览 6提问于2020-04-24得票数 0
3回答
当我们的kafka主题中有多个分区时，分区重新平衡是一件常见的事情吗？
这并不一定意味着我们的应用程序中存在一些延迟或问题？
我一直在查看被撤销和重新分配的分区的日志。
浏览 1提问于2018-11-16得票数 1
1回答
我有一个应用程序，我想在多个实例中打开，我需要每个实例来接收message.Actually，只有当实例有不同的组it时，每个实例才会接收消息。我放置了比实际实例更多的分区，并且work.The应用程序并不同时充当消费者/生产者。代码：
@EnableKafka
@Configuration
public class KafkaProducerConfigForDepartment {
@Value(value = "${kafka.bootstrapAddress}")
private String bootstrapAddress;
@Bean
浏览 2提问于2022-11-08得票数 0
我在Kafka主题中有两个消息(假设偏移量1和2)，这两个消息被放在同一个分区中(假设是p1)。
我的消费者应用是这样的：
我的消费者正在获取消息1(偏移量为1)，并向Kafka发送手动提交信号，然后等待5秒。
我的预期是因为提交信号发送到了kafka，而我的线程1正在等待5秒，而另一个使用者线程应该从分区1获取消息2，并在单独的线程中处理它。
然而，它不是这样工作的。它正在一个接一个地处理。只有在线程1完成5秒后，它才会从主题中提取第二条消息。
注意:我已经确保将不并行消费者的数量设置为多个(在我的示例中为5，最大消费者池大小为10)。
我做错什么了吗？有没有人遇到过类似的问题？如果是
浏览 16提问于2021-03-15得票数 1
1回答
在我们的系统中，我们在tomcat应用程序中使用旧版本的kafka (0.9.0.1)和旧的scala消费者API。大多数时候一切都很好，但有时当消费者运行的服务器被应用程序中的其他任务大量使用时，消费者就会变得没有反应，这就像预期的那样触发了重新平衡，消费者被从其分区中删除，其他消费者被使用。
我的问题是，当消费者重新启动时，是否有一种简单的方法来重新注册自己？
我知道老客户在Zookeeper中存储分区客户详细信息，并认为我们可以有一个任务，定期检查我们的客户是否在那里注册，如果没有，就重新启动客户，但我不确定我们到底应该在那里检查什么。谁能告诉我一些关于kafka在zookeeper中存
浏览 0提问于2018-10-03得票数 0
我使用的是kafka_image=wurstmeister/kafka
zookeeper_version=3.4.14 kafka_version=2.12-2.4.0
C#客户端: Confluent kafka v1.2.0
我们使用3个代理，1个动物园管理员集群作为一个pat部署，我们使用它来停止所有的经纪人，动物园管理员，生产者，消费者和删除卡夫卡日志文件，并启动消费者首先将启动经纪人和动物园管理员。在这个过程中，消费者会陷入困境，即使是活着的，也不会选择任何消息。如果我重新启动消费者，它就会开始选择
浏览 0提问于2020-02-05得票数 2
我执行shell脚本kafka-topics.sh将分区数从3改为10。
./bin/kafka-topics.sh --zookeeper xxx.xx.xxx.xx:2181/my-broker-0 --alter --topic target-topic-example-1 --partitions 10
我一执行这个命令，Kafka集群就出现了错误。
[2018-11-20 14:45:29,954] INFO Created log for partition [target-topic-example-1,5] in /data/kafka/brokers with propert
浏览 0提问于2018-11-20得票数 0
我正在设置kafka集群，想知道auto.leader.rebalance.enable属性的推荐值是多少？kafka默认值为true。是否可以将其保留为默认值？kafka默认值会导致问题吗？
我使用的Kafka版本是2.5.0
浏览 0提问于2020-07-25得票数 1
假设我有一个多线程的kafkaStreams应用程序。它正在运行，并且所有分区都已分配，因此没有线程分配超过1个分区。
现在，这个应用程序上的CPU上升了很多，我的自动定标器启动了另一个实例。我的新实例会被分配任何分区吗？或者kafka会说，“我把这些分区尽可能地分散开了，感谢新的实例，但我只会在现有的实例崩溃的情况下使用它”？
假设答案是，是的，kafka不会重新平衡，有什么解决方案可以让kafka在实例之间平衡
浏览 1提问于2018-04-14得票数 0
我使用手动kafka commit，将属性enable.auto.commit设置为false，同时初始化Kafka消费者，并在接收和处理消息后手动调用kafka commit。
然而，因为在我的消费者中处理消息是很耗时的，所以我得到了Exception with message "error": "Broker: Group rebalance in progress"
原因是重新平衡超时后提交被拒绝，并出现此错误。现在，恢复操作是退出并重新实例化进程，这将再次触发重新平衡和分区分配。另一种方法是捕获此异常，然后照常继续，只有在poll()调用被阻塞直到重新平
浏览 15提问于2020-05-19得票数 0
我们已经在RHEL 7.6 linux版本上安装了3台kafka代理机器。
卡夫卡版本是2.7.x
每个kafka都有8个Jbod磁盘，如下所示( df -h详细信息)
df -h
/dev/sdc 1.7T 929G 748G 56% /kafka/kafka_logs2
/dev/sdd 1.7T 950G 727G 57% /kafka/kafka_logs3
/dev/sde 1.7T 999G 678G 60% /kafka/kafka_logs4
/
浏览 3提问于2021-10-26得票数 0
我正在尝试更改Kafka主题配置，我可以通过更改配置来保留我的数据和偏移数据吗？我想知道当我将分区编号1改为3(或将压缩类型gzip改为lz4)或在删除后重新创建时，消息和偏移量信息是否被删除？
浏览 12提问于2021-01-29得票数 0
我有一个使用kafka streams的spring boot应用程序(kafka docker镜像：wurstmeister/kafka:2.12-2.1.1，kafka依赖：org.apache.kafka:kafka-streams:2.4.1)。在应用程序启动期间，我检查是否创建了主题my-topic，如果没有- application创建了它。在该应用程序创建KTable之后，如下所示：
streamsBuilder.table("my-topic", Consumed.with(Serdes.String(), Serdes.String()), Materiali
浏览 1提问于2020-05-22得票数 0
为什么在启动Kafka流应用程序时，我会断断续续地点击下面的警告消息？
这不仅仅是一个警告，因为它淹没了应用程序日志& Kafka应用程序没有启动。
通常情况下，当我重新部署时，它就能工作了。
[my-app-0 my-app] 2020-03-25 14:00:12.931 WARN 1 --- [-StreamThread-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=my-app-b8f0b2a0-271b-4499-85bd-9e22d4a8b4b1-StreamThread-1-consu
浏览 1提问于2020-03-25得票数 7
回答已采纳
我有一个用例，在该用例中，我必须测试具有高no消费者组的Kafka (3节点)集群的饱和点。(为我们的生产用例寻找饱和点)生产者ack=all。我创建了许多超过10000的消费者组，没有问题( no刚刚创建的消费者组不消费)。
于是我开始了负载测试，我用3个复制因子创建了3个主题(1个分区)，每个代理都是一个主题的领导者(我通过kafka- topics确定了这一点)。
我计划不断为每个主题生成4.5MBps，并将消费者组从零、100 45大小的45条记录增加到一个主题。
当我为集群中没有使用者组生成数据时，生产者响应时间仅为2ms/记录。
每记录100个消费者群体，花费7毫秒。
当某个主题的
浏览 2提问于2022-01-05得票数 1
回答已采纳
我们有Kafka集群版本2.7
每台机器都有一个磁盘，如下所示
/dev/sdb 82% /var/kafka/data1
所有主题都在/var/kafka/data1 1下面。
我们决定向每个kafka添加7个磁盘，如下所示
我们在每个新磁盘上创建了XFS文件系统，包括挂载(和fstab配置)。
具体如下：
/dev/sdb 82% /var/kafka/data1
/dev/sdc 1% /var/kafka/data2
/dev/sdd
浏览 1提问于2022-01-18得票数 0
回答已采纳
谢谢你的耐心。
在线为主题添加分区后，kafka消费者停止读取消息，不会抛出异常。消费者只是阻塞。每次我们都必须重启消费者。我认为这是不合理的，我找不到任何关于它的文档。
此外，当处理消息时出现错误时，消费者线程将不会恢复。我们的消费者读取消息并将其插入到MySql中。一旦网络出现故障，消费者就无法连接到MySql，然后它就会阻塞并停止读取消息，直到我们重新启动它。
添加分区时，旧数据和新数据会发生什么情况？文档()上写着：
“请注意，分区的一个用例是对数据进行语义分区，添加分区不会改变现有数据的分区，因此如果消费者依赖于该分区，这可能会干扰他们。也就是说，如果数据是按散列(键)% numbe
浏览 57提问于2016-03-20得票数 4
1回答
如果我创建一个带有组consumerGroup的消费者C1来读取topic A中的数据，会怎么样呢？一段时间后，在同一组中创建消费者C2，以从主题B读取数据。
创建消费者C2是否会触发重新平衡？作为一个更普遍的问题，kafka将在什么时候执行重新平衡？
浏览 1提问于2017-02-22得票数 3
回答已采纳
我感觉在Java库重新构建KafkaStreams时，标点符号能够访问本地存储数据，重放changelog主题中的项目 让我们来描述一下哪个是我们的场景:我有一个KafkaStreams应用程序A(用Java语言编写)，在这个应用程序中，我们使用一个主题“into topic”，将一些对象放入一个本地存储中，并使用一个时间戳加上一些唯一的id作为键，如下所示： key -> 201906122345_ds243132eddsfs value ->对象 然后我们有一个标点符号，从这个商店读取任何一分钟，检索从"“到现在表示为yyyymmddhhMM (so 20190612
浏览 28提问于2019-06-18得票数 0
1回答
我使用的是由卡夫卡连接而成的Debezium。Kafka Connect公开了一些需要创建的主题：
OFFSET_STORAGE_TOPIC在运行Kafka服务时需要这个环境变量。将此设置为Kafka主题的名称，在该主题中，组存储连接器中的Kafka服务会偏移。主题应该有许多分区，高度复制(例如，3x或更多)，并且应该配置为压缩。
STATUS_STORAGE_TOPIC在运行Kafka服务时应该提供这个环境变量。将此设置为Kafka主题的名称，在该主题中，组存储连接器状态中的Kafka服务。主题可以有多个分区，应该被高度复制(例如，3x或更多)，并且应该配置为压缩。
是否有人对这些主题有任何
浏览 4提问于2022-05-03得票数 0
回答已采纳
1回答
我有一个与事件驱动系统中的棘手情况相关的问题，我想请教一下。情况是这样的：
在我们的系统中，我使用redis作为memcached数据库，使用kafkaa作为消息队列。为了提高redis的性能，我使用lua scripting来处理数据，同时将事件推入redis的阻塞列表中。然后，将有一个过程来挑选该阻止列表中的redis事件，并将它们移动到kafka。因此，在这个过程中，有3个步骤：
1)读取redis列表中的事件
2)批量生产成kafka
3)删除redis中对应的事件
不幸的是，如果进程在2到3之间终止，意味着在kafka中产生所有事件后，它不会删除redis中的相应事件，那么在该进程重
浏览 0提问于2018-09-30得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
