title: Spring Cloud Stream Kafka - Serde class not found: org.apache.kafka.common.serialization.Serde$StringSerde
body: <p>I am trying to build a simple Kafka Streams application using Spring Cloud Stream framework. I can connect to the stream to push raw data for processing. But when I try to process the stream for event counting by key, I get <code>Serde class not found: org.apache.kafka.common.serialization.Serde$StringSerde</code> exception when I run the application. I checked in my project included libraries and I could locate the <code>Serde</code> class, it's not missing. I am not sure why at run time it is not being loaded!</p>
  
  <p>Below are my source files.</p>
  
  <p><code>com.pgp.learn.kafka.analytics.AnalyticsApplication</code></p>
  
  <pre><code>package com.pgp.learn.kafka.analytics;
  
  import com.pgp.learn.kafka.analytics.model.PageViewEvent;
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.kafka.common.serialization.Serde;
  import org.apache.kafka.common.serialization.Serdes;
  import org.apache.kafka.streams.KeyValue;
  import org.apache.kafka.streams.kstream.*;
  import org.springframework.boot.ApplicationArguments;
  import org.springframework.boot.ApplicationRunner;
  import org.springframework.boot.SpringApplication;
  import org.springframework.boot.autoconfigure.SpringBootApplication;
  import org.springframework.cloud.stream.annotation.EnableBinding;
  import org.springframework.cloud.stream.annotation.Input;
  import org.springframework.cloud.stream.annotation.StreamListener;
  import org.springframework.kafka.support.KafkaHeaders;
  import org.springframework.messaging.Message;
  import org.springframework.messaging.MessageChannel;
  import org.springframework.messaging.handler.annotation.SendTo;
  import org.springframework.messaging.support.MessageBuilder;
  import org.springframework.stereotype.Component;
  
  import java.util.Arrays;
  import java.util.List;
  import java.util.Random;
  import java.util.concurrent.Executors;
  import java.util.concurrent.TimeUnit;
  
  @SpringBootApplication
  @EnableBinding(AnalyticsBinding.class)
  public class  AnalyticsApplication {
  
      public static void main(String[] args) {
          SpringApplication.run(AnalyticsApplication.class, args);
      }
  
      @Component
      public static class PageViewEventSource implements ApplicationRunner {
          private final MessageChannel pageViewOut;
          private final Log log = LogFactory.getLog(getClass());
  
          public PageViewEventSource(AnalyticsBinding binding) {
              this.pageViewOut = binding.pageViewsOut();
          }
  
          @Override
          public void run(ApplicationArguments args) throws Exception {
              List&lt;String&gt; names = Arrays.asList("Peter", "Tom", "Ady", "Nency", "George", "Kevin", "Chelsey", "Thomas");
              List&lt;String&gt; pages = Arrays.asList("About", "Contact", "Blogs", "Gallery", "Music", "Site-map", "News");
  
              Runnable runner = () -&gt; {
                  String rName = names.get(new Random().nextInt(names.size()));
                  String rPage = pages.get(new Random().nextInt(pages.size()));
  
                  PageViewEvent event = new PageViewEvent(rName, rPage, Math.random() &gt; .5 ? 10L : 1000L);
  
                  Message&lt;PageViewEvent&gt; message = MessageBuilder
                          .withPayload(event)
                          .setHeader(KafkaHeaders.MESSAGE_KEY, event.getUserId().getBytes())
                          .build();
  
                  try {
                      this.pageViewOut.send(message);
                      log.info("Sent: " + message.toString());
                  } catch (Exception e) {
                      log.error(e);
                  }
              };
  
              Executors.newScheduledThreadPool(1).scheduleAtFixedRate(runner, 1, 1, TimeUnit.SECONDS);
          }
      }
  
      @Component
      public static class PageViewEventProcessor {
          private final Log log = LogFactory.getLog(getClass());
  
          @StreamListener
          @SendTo(AnalyticsBinding.PAGE_COUNT_OUT)
          public static KStream&lt;String, Long&gt; process(
                  @Input(AnalyticsBinding.PAGE_VIEW_IN_CHANNEL) KStream&lt;String, PageViewEvent&gt; events) {
              return events
                      .filter((s, pageViewEvent) -&gt; pageViewEvent.getDuration() &gt; 10)
                      .map((s, pageViewEvent) -&gt; new KeyValue&lt;&gt;(pageViewEvent.getPage(), 0L))
                      .groupByKey()
                      //Also tried using below line, but no luck
  //.groupByKey(Serialized.with(Serdes.String(), Serdes.Long()))
  
  .count(Materialized.as(AnalyticsBinding.PAGE_COUNT_MV))
                      .toStream();
          }
      }
  }
  </code></pre>
  
  <p><code>com.pgp.learn.kafka.analytics.AnalyticsBinding</code></p>
  
  <pre><code>package com.pgp.learn.kafka.analytics;
  
  import com.pgp.learn.kafka.analytics.model.PageViewEvent;
  import org.apache.kafka.streams.kstream.KStream;
  import org.springframework.cloud.stream.annotation.Input;
  import org.springframework.cloud.stream.annotation.Output;
  import org.springframework.messaging.MessageChannel;
  
  public interface AnalyticsBinding {
  
      String PAGE_VIEW_OUT_CHANNEL = "pvout";
      String PAGE_VIEW_IN_CHANNEL = "pvin";
      String PAGE_COUNT_MV = "pcmvc";
      String PAGE_COUNT_OUT = "pcout";
      String PAGE_COUNT_IN = "pcin";
  
      @Input (PAGE_VIEW_IN_CHANNEL)
      KStream&lt;String, PageViewEvent&gt; pageViewsIn();
  
      @Output (PAGE_VIEW_OUT_CHANNEL)
      MessageChannel pageViewsOut();
  
      @Output (PAGE_COUNT_OUT)
      KStream&lt;String, Long&gt; pageCountOut();
  
  }
  </code></pre>
  
  <p><code>application.properties</code></p>
  
  <pre><code># defaults
  spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.mms=1000
  spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serde$StringSerde
  spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serde$StringSerde
  spring.cloud.stream.kafka.binder.brokers=54.173.206.255
  #
  # page views out
  spring.cloud.stream.bindings.pvout.destination=pvs
  spring.cloud.stream.bindings.pvout.producer.header-mode=raw
  #
  # page views in
  spring.cloud.stream.bindings.pvin.destination=pvs
  spring.cloud.stream.bindings.pvin.consumer.header-mode=raw
  #
  # page count out
  spring.cloud.stream.bindings.pcout.destination=pcs
  spring.cloud.stream.bindings.pcout.producer.use-native-encoding=true
  spring.cloud.stream.kafka.streams.bindings.pcout.producer.keySerde=org.apache.kafka.common.serialization.Serdes$StringSerde
  spring.cloud.stream.kafka.streams.bindings.pcout.producer.valueSerde=org.apache.kafka.common.serialization.Serdes$LongSerde
  </code></pre>
  
  <p><code>pom.xml</code></p>
  
  <pre><code>&lt;?xml version="1.0" encoding="UTF-8"?&gt;
  &lt;project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"&gt;
      &lt;modelVersion&gt;4.0.0&lt;/modelVersion&gt;
      &lt;parent&gt;
          &lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
          &lt;artifactId&gt;spring-boot-starter-parent&lt;/artifactId&gt;
          &lt;version&gt;2.1.1.RELEASE&lt;/version&gt;
          &lt;relativePath/&gt; &lt;!-- lookup parent from repository --&gt;
      &lt;/parent&gt;
      &lt;groupId&gt;com.pgp.learn.kafka&lt;/groupId&gt;
      &lt;artifactId&gt;analytics&lt;/artifactId&gt;
      &lt;version&gt;0.0.1-SNAPSHOT&lt;/version&gt;
      &lt;name&gt;analytics&lt;/name&gt;
      &lt;description&gt;Demo project for Spring Boot&lt;/description&gt;
  
      &lt;properties&gt;
          &lt;java.version&gt;1.8&lt;/java.version&gt;
          &lt;spring-cloud.version&gt;Greenwich.RC2&lt;/spring-cloud.version&gt;
      &lt;/properties&gt;
  
      &lt;dependencies&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
              &lt;artifactId&gt;spring-boot-starter-web&lt;/artifactId&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.springframework.cloud&lt;/groupId&gt;
              &lt;artifactId&gt;spring-cloud-stream&lt;/artifactId&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.springframework.cloud&lt;/groupId&gt;
              &lt;artifactId&gt;spring-cloud-stream-binder-kafka&lt;/artifactId&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.springframework.cloud&lt;/groupId&gt;
              &lt;artifactId&gt;spring-cloud-stream-binder-kafka-streams&lt;/artifactId&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.springframework.kafka&lt;/groupId&gt;
              &lt;artifactId&gt;spring-kafka&lt;/artifactId&gt;
          &lt;/dependency&gt;
  
          &lt;dependency&gt;
              &lt;groupId&gt;org.projectlombok&lt;/groupId&gt;
              &lt;artifactId&gt;lombok&lt;/artifactId&gt;
              &lt;optional&gt;true&lt;/optional&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
              &lt;artifactId&gt;spring-boot-starter-test&lt;/artifactId&gt;
              &lt;scope&gt;test&lt;/scope&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.springframework.cloud&lt;/groupId&gt;
              &lt;artifactId&gt;spring-cloud-stream-test-support&lt;/artifactId&gt;
              &lt;scope&gt;test&lt;/scope&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.springframework.kafka&lt;/groupId&gt;
              &lt;artifactId&gt;spring-kafka-test&lt;/artifactId&gt;
              &lt;scope&gt;test&lt;/scope&gt;
          &lt;/dependency&gt;
      &lt;/dependencies&gt;
  
      &lt;dependencyManagement&gt;
          &lt;dependencies&gt;
              &lt;dependency&gt;
                  &lt;groupId&gt;org.springframework.cloud&lt;/groupId&gt;
                  &lt;artifactId&gt;spring-cloud-dependencies&lt;/artifactId&gt;
                  &lt;version&gt;${spring-cloud.version}&lt;/version&gt;
                  &lt;type&gt;pom&lt;/type&gt;
                  &lt;scope&gt;import&lt;/scope&gt;
              &lt;/dependency&gt;
          &lt;/dependencies&gt;
      &lt;/dependencyManagement&gt;
  
      &lt;build&gt;
          &lt;plugins&gt;
              &lt;plugin&gt;
                  &lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
                  &lt;artifactId&gt;spring-boot-maven-plugin&lt;/artifactId&gt;
              &lt;/plugin&gt;
          &lt;/plugins&gt;
      &lt;/build&gt;
  
      &lt;repositories&gt;
          &lt;repository&gt;
              &lt;id&gt;spring-milestones&lt;/id&gt;
              &lt;name&gt;Spring Milestones&lt;/name&gt;
              &lt;url&gt;https://repo.spring.io/milestone&lt;/url&gt;
          &lt;/repository&gt;
      &lt;/repositories&gt;
  
  &lt;/project&gt;
  </code></pre>
  
  <p><strong>Exception while Running</strong></p>
  
  <pre><code>java.lang.IllegalStateException: java.lang.IllegalStateException: Serde class not found: 
      at org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsStreamListenerSetupMethodOrchestrator.adaptAndRetrieveInboundArguments(KafkaStreamsStreamListenerSetupMethodOrchestrator.java:308) ~[spring-cloud-stream-binder-kafka-streams-2.1.0.RC4.jar:2.1.0.RC4]
      at org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsStreamListenerSetupMethodOrchestrator.orchestrateStreamListenerSetupMethod(KafkaStreamsStreamListenerSetupMethodOrchestrator.java:164) ~[spring-cloud-stream-binder-kafka-streams-2.1.0.RC4.jar:2.1.0.RC4]
      at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.doPostProcess(StreamListenerAnnotationBeanPostProcessor.java:195) ~[spring-cloud-stream-2.1.0.RC4.jar:2.1.0.RC4]
      at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.lambda$postProcessAfterInitialization$0(StreamListenerAnnotationBeanPostProcessor.java:167) ~[spring-cloud-stream-2.1.0.RC4.jar:2.1.0.RC4]
      at java.lang.Iterable.forEach(Iterable.java:75) ~[na:1.8.0_161]
      at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.injectAndPostProcessDependencies(StreamListenerAnnotationBeanPostProcessor.java:285) ~[spring-cloud-stream-2.1.0.RC4.jar:2.1.0.RC4]
      at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.afterSingletonsInstantiated(StreamListenerAnnotationBeanPostProcessor.java:105) ~[spring-cloud-stream-2.1.0.RC4.jar:2.1.0.RC4]
      at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:863) ~[spring-beans-5.1.3.RELEASE.jar:5.1.3.RELEASE]
      at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:863) ~[spring-context-5.1.3.RELEASE.jar:5.1.3.RELEASE]
      at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:546) ~[spring-context-5.1.3.RELEASE.jar:5.1.3.RELEASE]
      at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:142) ~[spring-boot-2.1.1.RELEASE.jar:2.1.1.RELEASE]
      at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:775) [spring-boot-2.1.1.RELEASE.jar:2.1.1.RELEASE]
      at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:397) [spring-boot-2.1.1.RELEASE.jar:2.1.1.RELEASE]
      at org.springframework.boot.SpringApplication.run(SpringApplication.java:316) [spring-boot-2.1.1.RELEASE.jar:2.1.1.RELEASE]
      at org.springframework.boot.SpringApplication.run(SpringApplication.java:1260) [spring-boot-2.1.1.RELEASE.jar:2.1.1.RELEASE]
      at org.springframework.boot.SpringApplication.run(SpringApplication.java:1248) [spring-boot-2.1.1.RELEASE.jar:2.1.1.RELEASE]
      at com.pgp.learn.kafka.analytics.AnalyticsApplication.main(AnalyticsApplication.java:35) [classes/:na]
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_161]
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_161]
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_161]
      at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_161]
      at org.springframework.boot.maven.AbstractRunMojo$LaunchRunner.run(AbstractRunMojo.java:558) [spring-boot-maven-plugin-2.1.1.RELEASE.jar:2.1.1.RELEASE]
      at java.lang.Thread.run(Thread.java:748) [na:1.8.0_161]
  Caused by: java.lang.IllegalStateException: Serde class not found: 
      at org.springframework.cloud.stream.binder.kafka.streams.KeyValueSerdeResolver.getKeySerde(KeyValueSerdeResolver.java:177) ~[spring-cloud-stream-binder-kafka-streams-2.1.0.RC4.jar:2.1.0.RC4]
      at org.springframework.cloud.stream.binder.kafka.streams.KeyValueSerdeResolver.getInboundKeySerde(KeyValueSerdeResolver.java:74) ~[spring-cloud-stream-binder-kafka-streams-2.1.0.RC4.jar:2.1.0.RC4]
      at org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsStreamListenerSetupMethodOrchestrator.adaptAndRetrieveInboundArguments(KafkaStreamsStreamListenerSetupMethodOrchestrator.java:246) ~[spring-cloud-stream-binder-kafka-streams-2.1.0.RC4.jar:2.1.0.RC4]
      ... 22 common frames omitted
  Caused by: java.lang.ClassNotFoundException: org.apache.kafka.common.serialization.Serde$StringSerde
      at java.net.URLClassLoader.findClass(URLClassLoader.java:381) ~[na:1.8.0_161]
      at java.lang.ClassLoader.loadClass(ClassLoader.java:424) ~[na:1.8.0_161]
      at java.lang.ClassLoader.loadClass(ClassLoader.java:357) ~[na:1.8.0_161]
      at java.lang.Class.forName0(Native Method) ~[na:1.8.0_161]
      at java.lang.Class.forName(Class.java:348) ~[na:1.8.0_161]
      at org.apache.kafka.common.utils.Utils.loadClass(Utils.java:333) ~[kafka-clients-2.0.1.jar:na]
      at org.apache.kafka.common.utils.Utils.newInstance(Utils.java:322) ~[kafka-clients-2.0.1.jar:na]
      at org.springframework.cloud.stream.binder.kafka.streams.KeyValueSerdeResolver.getKeySerde(KeyValueSerdeResolver.java:171) ~[spring-cloud-stream-binder-kafka-streams-2.1.0.RC4.jar:2.1.0.RC4]
      ... 24 common frames omitted
  </code></pre>
  
  <p>Any help to find what I am doing wrong here? I am new to Kafka and Stream processing. So, please excuse me if I am doing any silly mistake.</p>