在broker重新启动很长时间后加载偏移量和元数据块KafkaConsumer

Question

我们有一个问题，有时调用‘轮询’方法的新KafkaConsumer挂起长达20到30分钟后，三个kafka经纪人中的一个得到重新启动！

我们使用的是3 broker kafka设置(0.9.0.1)。我们的消费者进程使用新的Java KafkaConsumer-API，并且我们将分配给特定的TopicPartition。

由于不同的原因我不能在这里展示真正的代码，但基本上我们的代码是这样工作的：

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.

我想知道是什么使得加载过程如此缓慢，该怎么做呢？

Answer 1

找到原因了。

代理的server.xml配置文件包含属性

log.cleaner.enable=false

（默认情况下，对于版本0.9.0.1，该属性为true）这意味着kafkas内部compacted__consumer_offsets主题实际上没有被压缩，因为禁用了log-cleaner。实际上，本主题的一些分区增长到了几千兆字节的大小，这就解释了当一个新的组协调器需要重新填充它的缓存时，读取所有消费者偏移数据所需要的时间。