AbstractMethodError with Spark 1.6.0 and Kafka 0.10.2

[dibbhatt]

AbstractMethodError with Spark 1.6.0 and Kafka 0.10.2

[dibbhatt]

Which version of the consumer you are running ? It is due to version of your Spark (1.6.0) and version in pom doesn't match. You can git clone the code and update the consumer pom to match your version and try. But using spark 1.6 you may see couple of compilation issue which are easy to solve.

Here are the steps you can try.

1. git clone the latest code.

2. modify pom.xml to match your kafka and spark version ( including scala version)

e.g.

**
<spark.version>1.6.0</spark.version>
<kafka.version>0.10.2.0</kafka.version>

  <groupId>org.apache.spark</groupId>
  <artifactId>spark-core_2.10</artifactId>**

and

  **<groupId>org.apache.spark</groupId>
  <artifactId>spark-streaming_2.10</artifactId>**

and

  **<groupId>org.apache.kafka</groupId>
  <artifactId>kafka_2.10</artifactId>**

2. As Spark 1.6 and 2.0 has some incompatible changes, you need to remove one Listener call back from
   consumer.kafka.ReceiverStreamListener.java

remove this import

import org.apache.spark.streaming.scheduler.StreamingListenerStreamingStarted;

and remove this call back

**@Override
public void onStreamingStarted(StreamingListenerStreamingStarted arg0) {
}**

3. Spark 1.6 and 2.0 has another incompatibility for return type of PairFlatMapFunction. So you need to modify this file consumer.kafka.PartitionOffsetPair.java

change return type of call method

public Iterator<Tuple2<Integer, Long>> call(Iterator<MessageAndMetadata> it)

to

public Iterable<Tuple2<Integer, Long>> call(Iterator<MessageAndMetadata> it)

And change the return type from

    return kafkaPartitionToOffsetList.iterator();

to

    **return kafkaPartitionToOffsetList;**

That's it. Build the consumer and you should be all set to use it for Spark 1.6 and Kafka 0.10.2

Let me know if you face any issues.

Dibyendu

[dibbhatt]

Or another option is use consumer version 1.0.9. That will work with Spark 1.6

	<dependency>
		<groupId>dibbhatt</groupId>
		<artifactId>kafka-spark-consumer</artifactId>
		**<version>1.0.9</version>**
	</dependency>

[dibbhatt]

Here is the V 1.0.9 READ ME
https://github.com/dibbhatt/kafka-spark-consumer/tree/117f98ccf02ad4f6e5a8b8918b5db097e7d3a3d4

[MLNW]

Thank you for your quick response!

I used your first approach and modified the latest code to use my versions of Kafka, Spark and Scala. Seems to work.

I will do some more extensive testing during this week. If I find anything else I'll let you know.

Cheers!

[dibbhatt]

Perfect. Do let me know if you see any issues or need any help on tuning various knobs .

[LinMingQiang]

When spark job was submitted The system loaded the default jar of CDH(spark-assembly-1.6.0-cdh5.14.4-hadoop2.6.0-cdh5.14.4.jar)。The Kafka version is not 010。(0.9.0)

[dibbhatt]

Hi @LinMingQiang , in your Application pom, what version of jars you have specified ?

[LinMingQiang]

spark 1.6.0 kafka 0.10.0

[dibbhatt]

Whats the issue you see ? Is the streaming job not running ?