title: GCP dataproc - java.lang.NoClassDefFoundError: org/apache/kafka/common/serialization/ByteArraySerializer
body: <p>i'm trying to run a StructuredStreaming job on GCP DataProc, which reads from Kafka nd prints out the values. The code is giving error -&gt; java.lang.NoClassDefFoundError: org/apache/kafka/common/serialization/ByteArraySerializer</p>
  <p>Here is the code:</p>
  <pre><code>import sys, datetime, time, os
  from pyspark.sql.functions import col, rank, dense_rank, to_date, to_timestamp, format_number, row_number, lead, lag,monotonically_increasing_id
  from pyspark.sql import SparkSession, Window
  from confluent_kafka import Producer
  from google.cloud import storage
  
  spark = SparkSession.builder.appName('StructuredStreaming_VersaSase').getOrCreate()
  
  spark.sparkContext.setLogLevel(&quot;ERROR&quot;)
  
  kafkaBrokers='34.75.148.41:9094'
  topic = &quot;versa-sase&quot;
  # bootstrap.servers=my-cluster-lb-ssl-cert-kafka-bootstrap:9093
  security_protocol=&quot;SSL&quot;
  
  
  client = storage.Client()
  print(&quot; client &quot;, client)
  
  bucket = client.get_bucket('ssl-certs-karan')
  print(&quot; bucket &quot;, bucket)
  
  allblobs = bucket.list_blobs()
  print(&quot; allblobs -&gt; &quot;, allblobs)
  
  for b in allblobs:
      print(&quot; b -&gt; &quot;, b)
  
  blob_ssl_truststore_location = bucket.get_blob('ca.p12')
  print(&quot; blob_ssl_truststore_location.name &quot;, blob_ssl_truststore_location.name)
  blob_ssl_truststore_location.download_to_filename(blob_ssl_truststore_location.name)
  
  ssl_truststore_location=blob_ssl_truststore_location.name
  print(&quot; type - blob_ssl_truststore_location &quot;, type(blob_ssl_truststore_location))
  ssl_truststore_password=&quot;NAvqbh5c9fB4&quot;
  
  blob_ssl_keystore_location = bucket.get_blob('dataproc-versa-sase.p12')
  print(&quot; blob_ssl_keystore_location.name &quot;, blob_ssl_keystore_location.name)
  blob_ssl_keystore_location.download_to_filename(blob_ssl_keystore_location.name)
  ssl_keystore_location=blob_ssl_keystore_location.name
  ssl_keystore_password=&quot;jBGsWrBv7258&quot;
  consumerGroupId = &quot;versa-sase-grp&quot;
  checkpoint = &quot;gs://ss-checkpoint/&quot;
  
  print(&quot; SPARK.SPARKCONTEXT -&gt; &quot;, spark.sparkContext)
  
  
  
  df = spark.read.format('kafka')\
      .option(&quot;kafka.bootstrap.servers&quot;,kafkaBrokers)\
      .option(&quot;kafka.security.protocol&quot;,&quot;SSL&quot;) \
      .option(&quot;kafka.ssl.truststore.location&quot;,ssl_truststore_location) \
      .option(&quot;kafka.ssl.truststore.password&quot;,ssl_truststore_password) \
      .option(&quot;kafka.ssl.keystore.location&quot;, ssl_keystore_location)\
      .option(&quot;kafka.ssl.keystore.password&quot;, ssl_keystore_password)\
      .option(&quot;subscribe&quot;, topic) \
      .option(&quot;kafka.group.id&quot;, consumerGroupId)\
      .option(&quot;startingOffsets&quot;, &quot;earliest&quot;) \
      .load()
  #
  
  
  query = df.selectExpr(&quot;CAST(value AS STRING)&quot;) \
      .write \
      .format(&quot;console&quot;) \
      .option(&quot;numRows&quot;,100)\
      .option(&quot;checkpointLocation&quot;, checkpoint) \
      .option(&quot;outputMode&quot;, &quot;complete&quot;)\
      .save(&quot;output&quot;)
  
  # query.awaitTermination()
  </code></pre>
  <h1>Command to launch the job on Dataproc cluster :</h1>
  <pre><code>gcloud dataproc jobs submit pyspark \
  StructuredStreaming_Kafka_GCP-Batch-feb1.py --cluster=dataproc-ss-poc 
  --jars=gs://spark-jars-karan/spark-sql-kafka-0-10_2.12-3.1.2.jar     
  --region=us-central1
  
  </code></pre>
  <p>Error :</p>
  <pre><code> SPARK.SPARKCONTEXT -&gt;  &lt;SparkContext master=yarn appName=StructuredStreaming_VersaSase&gt;
  Traceback (most recent call last):
    File &quot;/tmp/b87ff69307344e2db5b43f4a73c377cf/StructuredStreaming_Kafka_GCP-Batch-feb1.py&quot;, line 49, in &lt;module&gt;
      df = spark.read.format('kafka')\
    File &quot;/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/readwriter.py&quot;, line 210, in load
    File &quot;/usr/lib/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py&quot;, line 1304, in __call__
    File &quot;/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/utils.py&quot;, line 111, in deco
    File &quot;/usr/lib/spark/python/lib/py4j-0.10.9-src.zip/py4j/protocol.py&quot;, line 326, in get_return_value
  py4j.protocol.Py4JJavaError: An error occurred while calling o69.load.
  : java.lang.NoClassDefFoundError: org/apache/kafka/common/serialization/ByteArraySerializer
      at org.apache.spark.sql.kafka010.KafkaSourceProvider$.&lt;init&gt;(KafkaSourceProvider.scala:556)
      at org.apache.spark.sql.kafka010.KafkaSourceProvider$.&lt;clinit&gt;(KafkaSourceProvider.scala)
      at org.apache.spark.sql.kafka010.KafkaSourceProvider.org$apache$spark$sql$kafka010$KafkaSourceProvider$$validateBatchOptions(KafkaSourceProvider.scala:336)
      at org.apache.spark.sql.kafka010.KafkaSourceProvider.createRelation(KafkaSourceProvider.scala:127)
      at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:355)
      at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:325)
      at org.apache.spark.sql.DataFrameReader.$anonfun$load$3(DataFrameReader.scala:307)
      at scala.Option.getOrElse(Option.scala:189)
      at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:307)
      at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:225)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.lang.reflect.Method.invoke(Method.java:498)
      at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
      at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
      at py4j.Gateway.invoke(Gateway.java:282)
      at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
      at py4j.commands.CallCommand.execute(CallCommand.java:79)
      at py4j.GatewayConnection.run(GatewayConnection.java:238)
      at java.lang.Thread.run(Thread.java:748)
  Caused by: java.lang.ClassNotFoundException: org.apache.kafka.common.serialization.ByteArraySerializer
      at java.net.URLClassLoader.findClass(URLClassLoader.java:387)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
  
  
  </code></pre>
  <p>I've checked the Spark version on the Dataproc cluster, and spark version - 3.1.2,and scala version - 2.12 .. so the version of the spark-sql jar being passed seems to be correct.
  Are there any other jars to be passed ?</p>
  <p>What needs to be done to fix/debug this issue ?</p>
  <p>tia !</p>