CODEWITHSUNDEEP
apache-sparkapache-kafkaspark-streaming
Gurubg
Gurubg

Reputation: 83

spark streaming with kafka how to handle handle driver exceptions

I have a SPARK streaming application which reads continuous incoming data from a kafka direct stream.

This is my setup 

   spark-core_2.10 - 2.1.1
   spark-streaming_2.10 - 2.1.1
   spark-streaming-kafka-0-10_2.10 - 2.0.0
   kafka_2.10 - 0.10.1.1

It runs pretty well most of the times, but sometimes when my other application which is writing to kafka restarts , I get the below errors . 

 WARN NetworkClient: Error while fetching metadata with correlation id 139022 : {topic4908100105=LEADER_NOT_AVAILABLE}
 18/05/24 11:59:33 WARN NetworkClient: Error while fetching metadata with correlation id 139022 : {topic000001=LEADER_NOT_AVAILABLE}
.
.
.

 ERROR JobScheduler: Error generating jobs for time 1527163130000 ms
 java.lang.IllegalArgumentException: requirement failed: numRecords must not be negative

I saw other posts in SO and people recommended to restart the kafka or spark application to recover from this error. But my application has to run continuously on its on without any manual intervention !

Is there any way to handle this exception in my Spark consumer class ( java) and restart the streaming or application ?

Upvotes: 0

Views: 877

Answers (1)

Ravikumar
Ravikumar

Reputation: 1131

WARN NetworkClient: Error while fetching metadata with correlation id 139022 : {topic4908100105=LEADER_NOT_AVAILABLE} 18/05/24 11:59:33 WARN NetworkClient: Error while fetching metadata with correlation id 139022 : {topic000001=LEADER_NOT_AVAILABLE}

This is Kafka issue, for each partition leader and followers are there. The request served by the leader. If leader is not available because of zookeeper issue, Kafka will throw this error.

You need to fix the Kafka issue to receive the data. However you can handle exception scenarios by using the following configuration properties.

  • --conf spark.yarn.maxAppAttempts=10
  • --conf spark.yarn.am.attemptFailuresValidityInterval=1h
  • --conf spark.yarn.max.executor.failures=10
  • --conf spark.yarn.executor.failuresValidityInterval=1h

Upvotes: 0

Related Questions