我们遇到的问题是，在三分之一的kafka经纪人重新启动后，有时会调用新KafkaConsumer的'poll'方法，因为 20 to 30 Minutes 就会挂起！
我们正在使用3经纪人kafka设置（0.9.0.1） . 我们的Consumer-Processes使用新的Java KafkaConsumer-API，我们将分配给特定的TopicPartitions .
由于不同的原因我不能在这里显示真正的代码，但基本上我们的代码的工作方式如下：
Properties consumerProps=loadConsumerProperties();
// bootstrap.servers=<IP1>:9092,<IP2>:9092,<IP3>:9092
// group.id="consumer_group_gwbc2
// enable.auto.commit=false
// auto.offset.reset=latest
// session.timeout.ms=30000
// key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
// value.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer
KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(consumerProps);
consumer.assign(Arrays.asList(new TopicPartition("someTopic",0)));
while (true) {
// THIS CALL sometimes blocks for a very long Time after a broker restart
ConsumerRecords<String, byte[]> records = kafkaConsumer.poll(200);
Iterator<ConsumerRecord<String, byte[]>> recordIter = records.iterator();
while (recordIter.hasNext()) {
ConsumerRecord<String, byte[]> record = recordIter.next();
// Very fast, actually just sending a UDP Paket via Netty.
processRecord(record);
if (lastCommitHappendFiveOrMoreSecondsAgo()) {
kafkaConsumer.commitAsync();
}
}
}
kafka-topics.sh描述 __consumer_offsets 主题如下
Topic:__consumer_offsets PartitionCount:50
ReplicationFactor:3 Configs:segment.bytes=104857600,
cleanup.policy=compact,compression.type=uncompressed
重新启动的代理的server.log显示从__consumer_offsets主题的特定分区加载偏移需要很长时间（在这种情况下大约需要22分钟） . 这与消费者的“民意调查”呼叫被阻止的时间相关 .
[2016-07-25 16:02:40,846] INFO [Group Metadata Manager on Broker 1]: Loading offsets and group metadata from [__consumer_offsets,15] (kafka.coordinator.GroupMetadataManager)
[2016-07-25 16:25:36,697] INFO [Group Metadata Manager on Broker 1]: Finished loading offsets from [__consumer_offsets,15] in 1375851 milliseconds.
我想知道是什么让加载过程如此缓慢以及可以做些什么呢！
1 回答
找到原因 .
我们的代理的server.xml配置文件包含该属性
（默认情况下，此属性从版本0.9.0.1开始为true）这意味着kafkas内部压缩 __consumer_offsets 主题实际上并未压缩，因为禁用了日志清理程序 . 实际上，该主题的某些分区增长到几千兆字节，这解释了当新的组协调器需要重新填充其缓存时，读取所有使用者偏移数据所需的时间 .