title: kafka streams AVRO registry with topology giving java.lang.NoClassDefFoundError for custom object
body: <p>AVRO schema is correct and the Deserialization is working fine with spring-kafka but not with kafka-streams.
  Getting noclassdeffounderror tough the jar is present in pom.</p>
  <pre><code>       config.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, &quot;SSL&quot;);
          config.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, keystoreLocation);
          config.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, keystorePassword);
          config.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, truststoreLocation);
          config.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, truststorePassword);
          config.put(StreamsConfig.APPLICATION_ID_CONFIG, &quot;test&quot;);
          config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
          config.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, consumerThreads);
          config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, &quot;earliest&quot;);
  
          config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
          config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
          config.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, avroRegistry);
  
          Serde&lt;CustomKey&gt; avroUserSerde = new SpecificAvroSerde&lt;&gt;();
          Serde&lt;CustomValue&gt; avroColorSerde = new SpecificAvroSerde&lt;&gt;();
  
          Map&lt;String, String&gt; configuration = new HashMap&lt;&gt;(3);
          configuration.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, avroRegistry);
  
          avroUserSerde.configure(configuration, true);
          avroColorSerde.configure(configuration, false);
  
          config.put(&quot;specific.avro.reader&quot;, true);
          Topology topology = new Topology();
  
          topology.addSource(&quot;SOURCE&quot;, avroUserSerde.deserializer(), avroColorSerde.deserializer(), topics);
          topology.addProcessor(&quot;PROCESSOR&quot;, () -&gt; new InputRecordProcessor(camelContext, context), &quot;SOURCE&quot;);
  
          KafkaStreams streams = new KafkaStreams(topology, config);
          streams.start();
  
  Exception Trace :
  
  ```
  Exception in thread &quot;imdm-streams-ijm-cdb-stg-22-27902e4d-66f6-45c7-b776-c0b812b0c885-StreamThread-6&quot; java.lang.NoClassDefFoundError: Could not initialize class com.optum.exts.eligibility.model.CustomObject
      at java.base/java.lang.Class.forName0(Native Method)
      at java.base/java.lang.Class.forName(Class.java:398)
      at org.apache.avro.util.ClassUtils.forName(ClassUtils.java:102)
      at org.apache.avro.util.ClassUtils.forName(ClassUtils.java:79)
      at org.apache.avro.specific.SpecificData.getClass(SpecificData.java:160)
  ```
  </code></pre>