title: Error implementing schema registry with kafka
body: <p>I'm trying to consume a record using spring-kata but getting the below exception during server startup. The producer is using the cdc via the confluent standalone with mysql . I'm able to consume it properly from the standalone consumer but consuming via spring-kafka seems to be a problem for me.</p>
  
  <pre><code>Caused by: java.lang.NoClassDefFoundError: io/confluent/common/config/ConfigException
      at java.lang.Class.forName0(Native Method) ~[na:1.8.0_131]
      at java.lang.Class.forName(Class.java:348) ~[na:1.8.0_131]
      at org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:709) ~[kafka-clients-2.1.1.jar:na]
      at org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:471) ~[kafka-clients-2.1.1.jar:na]
      at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:464) ~[kafka-clients-2.1.1.jar:na]
      at org.apache.kafka.common.config.AbstractConfig.&lt;init&gt;(AbstractConfig.java:62) ~[kafka-clients-2.1.1.jar:na]
      at org.apache.kafka.common.config.AbstractConfig.&lt;init&gt;(AbstractConfig.java:75) ~[kafka-clients-2.1.1.jar:na]
      at org.apache.kafka.clients.consumer.ConsumerConfig.&lt;init&gt;(ConsumerConfig.java:511) ~[kafka-clients-2.1.1.jar:na]
      at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:617) ~[kafka-clients-2.1.1.jar:na]
      at org.springframework.kafka.core.DefaultKafkaConsumerFactory.createKafkaConsumer(DefaultKafkaConsumerFactory.java:139) ~[spring-kafka-2.1.8.RELEASE.jar:2.1.8.RELEASE]
      at org.springframework.kafka.core.DefaultKafkaConsumerFactory.createKafkaConsumer(DefaultKafkaConsumerFactory.java:134) ~[spring-kafka-2.1.8.RELEASE.jar:2.1.8.RELEASE]
      at org.springframework.kafka.core.DefaultKafkaConsumerFactory.createConsumer(DefaultKafkaConsumerFactory.java:102) ~[spring-kafka-2.1.8.RELEASE.jar:2.1.8.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.&lt;init&gt;(KafkaMessageListenerContainer.java:425) ~[spring-kafka-2.1.8.RELEASE.jar:2.1.8.RELEASE]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer.doStart(KafkaMessageListenerContainer.java:259) ~[spring-kafka-2.1.8.RELEASE.jar:2.1.8.RELEASE]
      at org.springframework.kafka.listener.AbstractMessageListenerContainer.start(AbstractMessageListenerContainer.java:269) ~[spring-kafka-2.1.8.RELEASE.jar:2.1.8.RELEASE]
      at org.springframework.kafka.listener.ConcurrentMessageListenerContainer.doStart(ConcurrentMessageListenerContainer.java:164) ~[spring-kafka-2.1.8.RELEASE.jar:2.1.8.RELEASE]
      at org.springframework.kafka.listener.AbstractMessageListenerContainer.start(AbstractMessageListenerContainer.java:269) ~[spring-kafka-2.1.8.RELEASE.jar:2.1.8.RELEASE]
      at org.springframework.kafka.config.KafkaListenerEndpointRegistry.startIfNecessary(KafkaListenerEndpointRegistry.java:289) ~[spring-kafka-2.1.8.RELEASE.jar:2.1.8.RELEASE]
      at org.springframework.kafka.config.KafkaListenerEndpointRegistry.start(KafkaListenerEndpointRegistry.java:238) ~[spring-kafka-2.1.8.RELEASE.jar:2.1.8.RELEASE]
      at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:181) ~[spring-context-5.0.8.RELEASE.jar:5.0.8.RELEASE]
      ... 12 common frames omitted
  Caused by: java.lang.ClassNotFoundException: io.confluent.common.config.ConfigException
      at java.net.URLClassLoader.findClass(URLClassLoader.java:381) ~[na:1.8.0_131]
      at java.lang.ClassLoader.loadClass(ClassLoader.java:424) ~[na:1.8.0_131]
      at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335) ~[na:1.8.0_131]
      at java.lang.ClassLoader.loadClass(ClassLoader.java:357) ~[na:1.8.0_131]
      ... 32 common frames omitted
  </code></pre>
  
  <p>Dependencies</p>
  
  <pre><code>&lt;dependency&gt;
              &lt;groupId&gt;org.apache.avro&lt;/groupId&gt;
              &lt;artifactId&gt;avro&lt;/artifactId&gt;
              &lt;version&gt;1.8.2&lt;/version&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;io.confluent&lt;/groupId&gt;
              &lt;artifactId&gt;kafka-avro-serializer&lt;/artifactId&gt;
              &lt;version&gt;5.1.0&lt;/version&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.apache.kafka&lt;/groupId&gt;
              &lt;artifactId&gt;kafka-clients&lt;/artifactId&gt;
              &lt;version&gt;2.1.1&lt;/version&gt;
          &lt;/dependency&gt;
  </code></pre>
  
  <p>Config with the properties set</p>
  
  <pre><code>@Bean
      public KafkaListenerContainerFactory&lt;ConcurrentMessageListenerContainer&lt;GenericRecord, GenericRecord&gt;&gt; kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;GenericRecord, GenericRecord&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactory());
          //setting number of concurrent threads which will read from kafka topic,
          //basically it's a number of consumers in consumer group
          //it has concurrency 5, because number of partition in Kafka topic is 5, 1 thread per partition.
          factory.setConcurrency(5);
          //setting number of ms when to emit "notification" if kafka consumer is idle,
          //e.g. no more messages in kafka topic
          factory.getContainerProperties().setIdleEventInterval(3000L);
          return factory;
      }
  
      @Bean
      public ConsumerFactory&lt;GenericRecord, GenericRecord&gt; consumerFactory() {
          return new DefaultKafkaConsumerFactory&lt;&gt;(consumerConfigs());
      }
  
      @Bean
      public Map&lt;String, Object&gt; consumerConfigs() {
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName());
          props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, "true");
          props.put("schema.registry.url", "http://localhost:8081");
          props.put(ConsumerConfig.GROUP_ID_CONFIG, "avro");
  
          return props;
      }
  </code></pre>
  
  <p>Consumer which tries to read the message</p>
  
  <pre><code>@KafkaListener(topics = "mysql-foobar", groupId = "group_id")
      public void consume(ConsumerRecord&lt;String, Foobar&gt; message) {
  
          System.out.println("consumed message is value: "+message.value());
      }
  </code></pre>