一、问题描述

使用SparkStreaming读取kafka消息，采用Direct的方法读取，然后，在idea中调试时，报错：

19/12/08 17:11:31 INFO BlockManager: Initialized BlockManager: BlockManagerId(driver, 192.168.31.1, 61291, None)
Exception in thread "main" java.lang.ClassCastException: kafka.cluster.BrokerEndPoint cannot be cast to kafka.cluster.Broker
    at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$2$$anonfun$3$$anonfun$apply$6$$anonfun$apply$7.apply(KafkaCluster.scala:97)
    at scala.Option.map(Option.scala:146)
    at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$2$$anonfun$3$$anonfun$apply$6.apply(KafkaCluster.scala:97)
    at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$2$$anonfun$3$$anonfun$apply$6.apply(KafkaCluster.scala:94)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)

二、问题原因

竟然时因为pom文件中kafka_2.11和spark-streaming-kafka-0-8_2.11两个依赖有冲突导致。

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
      <version>2.1.0</version>
    </dependency>
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka_2.11</artifactId>
      <version>${kafka.version}</version>
    </dependency>

三、解决方法

注释kafka_2.11依赖

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
      <version>2.1.0</version>
    </dependency>
    <!--<dependency>-->
      <!--<groupId>org.apache.kafka</groupId>-->
      <!--<artifactId>kafka_2.11</artifactId>-->
      <!--<version>${kafka.version}</version>-->
    <!--</dependency>-->