java.lang.NoClassDefFoundError: io/confluent/kafka/serializers/AbstractKafkaSchemaSerDeConfig

[kothapet]

I was trying out this out in the python and getting this error. Running on EMR 6.6.0

22/06/28 20:39:38 INFO SchemaManagerFactory: Configuring new Schema Registry client of type ConfluentRegistryClient
Traceback (most recent call last):
  File "/home/hadoop/test.py", line 38, in <module>
    valueSchema = from_avro_abris_config({'schema.registry.url': SCHEMA_REGISTRY_URL}, KAFKA_READ_TOPIC, False)
  File "/home/hadoop/test.py", line 30, in from_avro_abris_config
    .usingSchemaRegistry(scala_map)
  File "/usr/lib/spark/python/lib/py4j-0.10.9.2-src.zip/py4j/java_gateway.py", line 1310, in __call__
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 111, in deco
  File "/usr/lib/spark/python/lib/py4j-0.10.9.2-src.zip/py4j/protocol.py", line 328, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o79.usingSchemaRegistry.
: java.lang.NoClassDefFoundError: io/confluent/kafka/serializers/AbstractKafkaSchemaSerDeConfig
        at java.lang.ClassLoader.defineClass1(Native Method)
        at java.lang.ClassLoader.defineClass(ClassLoader.java:756)
        at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
        at java.net.URLClassLoader.defineClass(URLClassLoader.java:473)

here is the code

from pyspark.sql import SparkSession
from pyspark.sql.column import Column, _to_java_column

spark = SparkSession.builder \
                    .config("spark.executor.heartbeatInterval","60s") \
                    .config("spark.network.timeout","61s") \
                    .appName("Python Spark SQL data source example") \
                    .getOrCreate()
sc = spark.sparkContext

def from_avro(col, config):
    
    jvm_gateway = sc._active_spark_context._gateway.jvm
    abris_avro = jvm_gateway.za.co.absa.abris.avro

    return Column(abris_avro.functions.from_avro(_to_java_column(col), config))

def from_avro_abris_config(config_map, topic, is_key):

    jvm_gateway = sc._active_spark_context._gateway.jvm
    print(f'jvm_gateway : {jvm_gateway}')
    scala_map = jvm_gateway.PythonUtils.toScalaMap(config_map)
    print(f'scala_map : {scala_map}')

    return jvm_gateway.za.co.absa.abris.config \
        .AbrisConfig \
        .fromConfluentAvro() \
        .downloadReaderSchemaByLatestVersion() \
        .andTopicNameStrategy(topic, is_key) \
        .usingSchemaRegistry(scala_map)

KAFKA_BOOTSTRAP_SERVER = "x.x.x.x:9092"
KAFKA_READ_TOPIC = "datagen.PURCHASES"
KAFKA_WRITE_TOPIC = "streams.PURCHASES_SUMMARY"

SCHEMA_REGISTRY_URL = "http://x.x.x.x:8081"

valueSchema = from_avro_abris_config({'schema.registry.url': SCHEMA_REGISTRY_URL}, KAFKA_READ_TOPIC, False)
keySchema   = from_avro_abris_config({'schema.registry.url': SCHEMA_REGISTRY_URL}, KAFKA_READ_TOPIC, True)

print(keySchema)            
print(valueSchema)            

spark.stop()

And I did add the following jars to the EMR 6.6.0

abris_2.12-6.1.1.jar
common-config-7.1.1.jar
common-utils-7.1.1.jar
commons-pool2-2.11.1.jar
kafka-avro-serializer-7.1.1.jar
kafka-clients-3.2.0.jar
kafka-schema-registry-7.1.1.jar
kafka-schema-registry-client-7.1.1.jar
spark-avro_2.12-3.2.0.jar
spark-sql-kafka-0-10_2.12-3.2.0.jar
spark-token-provider-kafka-0-10_2.12-3.2.0.jar
zkclient-0.11.jar

I see the AbstractKafkaSchemaSerDeConfig class is in kafka-avro-serializer-7.1.1.jar

[kothapet]

I guess the kafka-schema-serializer-7.1.1.jar was missing.. so all good here