解決spark streaming集成kafka時只能讀topic的其中一個分區數據的問題

1. 問題描述

我創建了一個名稱為myTest的topic，該topic有三個分區,在我的應用中spark streaming以direct方式連接kakfa，但是發現只能消費一個分區的數據，多次更換comsumer group依然如此。

2 環境配置

kafka集群環境，

主機	IP	操作系統	kakfa
node1	192.168.1.101	Centos 6.5	kafka_2.11-0.10.1.1
node2	192.168.1.102	Centos 6.5	kafka_2.11-0.10.1.1
node3	192.168.1.103	Centos 6.5	kafka_2.11-0.10.1.1

應用依賴：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>