Kafka-Spark Streaming-仅从1个分区读取数据

我有一个独立的Spark集群，它正在从kafka队列中读取数据。kafka队列有5个分区，spark仅处理来自其中一个分区的数据。我正在使用以下内容：

这是我的Maven依赖项：

    <dependencies>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
        <version>2.0.2</version>
    </dependency>   
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming_2.11</artifactId>
        <version>2.0.2</version>
    </dependency>
    <dependency>
        <groupId>kafka-custom</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>0.10.1.1</version>
    </dependency>   

我的kafka生产者是一个简单的生产者，仅将100条消息放在队列中：

    public void generateMessages() {

    // Define the properties for the Kafka Connection
    Properties props = new Properties();
    props.put("bootstrap.servers", kafkaBrokerServer); // kafka server
    props.put("acks", "all");
    props.put("retries", 0);
    props.put("batch.size", 16384);
    props.put("linger.ms", 1);
    props.put("buffer.memory", 33554432);
    props.put("key.serializer",
            "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer",
            "org.apache.kafka.common.serialization.StringSerializer");

    // Create a KafkaProducer using the Kafka Connection properties
    KafkaProducer<String, String> producer = new KafkaProducer<String, String>(
            props);
    for (int i = 0; i < 100; i++) {
        ProducerRecord<String, String> record = new ProducerRecord<>(kafkaTopic, "value-" + i);
        producer.send(record);
    }
    producer.close();

}

这是我的流传输作业中的主要代码：

    public void processKafka() throws InterruptedException {
    LOG.info("************ SparkStreamingKafka.processKafka start");

   // Create the spark application
    SparkConf sparkConf = new SparkConf();
    sparkConf.set("spark.executor.cores", "5");

    //To express any Spark Streaming computation, a StreamingContext object needs to be created. 
    //This object serves as the main entry point for all Spark Streaming functionality.
    //This creates the spark streaming context with a 'numSeconds' second batch size
    jssc = new JavaStreamingContext(sparkConf, Durations.seconds(sparkBatchInterval));


    //List of parameters
    Map<String, Object> kafkaParams = new HashMap<>();
    kafkaParams.put("bootstrap.servers", this.getBrokerList());
    kafkaParams.put("client.id", "SpliceSpark");
    kafkaParams.put("group.id", "mynewgroup");
    kafkaParams.put("auto.offset.reset", "earliest");
    kafkaParams.put("enable.auto.commit", false);
    kafkaParams.put("key.deserializer", StringDeserializer.class);
    kafkaParams.put("value.deserializer", StringDeserializer.class);

    List<TopicPartition> topicPartitions= new ArrayList<TopicPartition>();
    for(int i=0; i<5; i++) {
        topicPartitions.add(new TopicPartition("mytopic", i));
    }


    //List of kafka topics to process
    Collection<String> topics = Arrays.asList(this.getTopicList().split(","));


    JavaInputDStream<ConsumerRecord<String, String>> messages = KafkaUtils.createDirectStream(
            jssc,
            LocationStrategies.PreferConsistent(),
            ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams)
          );

    //Another version of an attempt
    /*
    JavaInputDStream<ConsumerRecord<String, String>> messages = KafkaUtils.createDirectStream(
        jssc,
        LocationStrategies.PreferConsistent(),
        ConsumerStrategies.<String, String>Assign(topicPartitions, kafkaParams)
      );
     */

    messages.foreachRDD(new PrintRDDDetails());


    // Start running the job to receive and transform the data 
    jssc.start();

    //Allows the current thread to wait for the termination of the context by stop() or by an exception
    jssc.awaitTermination();
}

PrintRDDDetails的调用方法具有以下内容：

    public void call(JavaRDD<ConsumerRecord<String, String>> rdd)
        throws Exception {

    LOG.error("--- New RDD with " + rdd.partitions().size()
            + " partitions and " + rdd.count() + " records");

}

似乎发生的事情是它仅从一个分区获取数据。我在kafka中确认有5个分区。当执行调用方法时，它会打印正确数量的分区，但仅打印1个分区中的记录-我从此简化代码中提取的进一步处理表明，它仅处理1个分区。