解决 spark streaming 集成 kafka 时只能读取每个 topic 的其中一个分区数据的问题

1. 问题描述

我创建了一个名称为 myTest 的 topic，该 topic 有三个分区,在我的应用中 spark streaming 以 direct 方式连接 kakfa，但是发现只能消费一个分区的数据，多次更换 comsumer group 依然如此。

2 环境配置

kafka 集群环境

应用依赖：spark 版本是 2.1.1、kakfa 版本是 0.10.1.1；
maven 依赖配置如下

<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.11</artifactId>
<version>2.1.1</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
<version>$2.1.1</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>0.10.1.1</version>
</dependency>

相关配置代码（Java）如下：

Map<String, Object> kafkaParams  =  new  HashMap<>();
kafkaParams.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092");
kafkaParams.put(ConsumerConfig.GROUP_ID_CONFIG, "myGroup");
kafkaParams.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
kafkaParams.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
kafkaParams.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
kafkaParams.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  
Set<String> topics  =  new  HashSet<String>(Arrays.asList("testTopic"));
JavaInputDStream<ConsumerRecord<Object, Object>> dStream  =  KafkaUtils.createDirectStream(
      jssc,
      LocationStrategies.PreferConsistent(),
      ConsumerStrategies.Subscribe(topics, kafkaParams));

3. 解决方案

经过查阅相关资料发现是由于 Kafka 0.10.1.1 的 bug 导致的。其实不仅仅是 0.10.1.1，另外 0.10.1.0 和 0.10.0.2 也有这个问题。详细描述参考 https://issues.apache.org/jira/browse/KAFKA-4547
最后我将 kafka 版本降到了 0.10.0.1，解决了这个问题。

<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>0.10.0.1</version>
</dependency>