title: Unable to set serde, producer and consumer properties per topic/binder level in spring cloud kafka
body: <p>I'm trying to bring up simple pub-sub application using spring cloud kafka binder. However I'm unable to set Serializer, DeSerialzer property and other producer and consumer properties in application.yml. I consistently get serialization/deserialization  error. Even kafka logs in spring boot project shows producer and consumer config still users ByteArraySerializer. Below is the code sameple.</p>
  
  <p><strong>pom.xml</strong></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 https://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.2.4.RELEASE&lt;/version&gt;
          &lt;relativePath/&gt; &lt;!-- lookup parent from repository --&gt;
      &lt;/parent&gt;
      &lt;groupId&gt;io.github.kprasad99.kafka&lt;/groupId&gt;
      &lt;artifactId&gt;kp-kafka-streams-example&lt;/artifactId&gt;
      &lt;version&gt;0.0.1-SNAPSHOT&lt;/version&gt;
      &lt;name&gt;kp-kafka-streams-example&lt;/name&gt;
      &lt;description&gt;Demo project for Spring Boot&lt;/description&gt;
  
      &lt;properties&gt;
          &lt;java.version&gt;11&lt;/java.version&gt;
          &lt;spring-cloud.version&gt;Hoxton.SR1&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;!-- &lt;dependency&gt;
              &lt;groupId&gt;org.apache.kafka&lt;/groupId&gt;
              &lt;artifactId&gt;kafka-streams&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-streams&lt;/artifactId&gt;
          &lt;/dependency&gt; --&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.kafka&lt;/groupId&gt;
              &lt;artifactId&gt;spring-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&lt;/artifactId&gt;
          &lt;/dependency&gt;
  
          &lt;dependency&gt;
              &lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
              &lt;artifactId&gt;spring-boot-devtools&lt;/artifactId&gt;
              &lt;scope&gt;runtime&lt;/scope&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-configuration-processor&lt;/artifactId&gt;
              &lt;optional&gt;true&lt;/optional&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;exclusions&gt;
                  &lt;exclusion&gt;
                      &lt;groupId&gt;org.junit.vintage&lt;/groupId&gt;
                      &lt;artifactId&gt;junit-vintage-engine&lt;/artifactId&gt;
                  &lt;/exclusion&gt;
              &lt;/exclusions&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;/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;/project&gt;
  </code></pre>
  
  <p><strong>Processor.java</strong></p>
  
  <pre><code>public interface Processor {
  
      String INPUT="k-msg-source";
      String OUTPUT="k-msg-sink";
  
      @Input(INPUT)
      SubscribableChannel input();
  
      @Output(OUTPUT)
      MessageChannel output();
  
  }
  </code></pre>
  
  <p><strong>KRest.java</strong></p>
  
  <p>@RestController
  public class KRest {</p>
  
  <pre><code>@Autowired
  private Processor processor;
  
  @GetMapping("/send")
  public ResponseEntity&lt;Void&gt; send(@RequestParam("key")String key, @RequestParam("msg") String text){
      processor.input().send(MessageBuilder.withPayload(Message.builder().text(text).build()).setHeader(KafkaHeaders.MESSAGE_KEY, key).build());
      return ResponseEntity.ok().build();
  }
  </code></pre>
  
  <p>}</p>
  
  <p>Message.java</p>
  
  <pre><code>@Data
  @NoArgsConstructor
  @AllArgsConstructor
  @Builder
  public class Message {
  
      private String text;
  }
  </code></pre>
  
  <p><strong>And finally application.yml</strong></p>
  
  <pre><code>spring:
    cloud:
      stream:
        bindings:
          k-msg-source:
            binder: kafka
            content-type: application/json
            destination: topic.kp.msg
            group: kp.msg.source
          k-msg-sink:
            binder: kafka
            content-type: application/json
            destination: topic.kp.msg
            group: kp.msg.sink
            producer:
              partition-count: 10
        binders:
          kafka:
            type: kafka
            environment:
              spring.cloud.stream.kafka.binder:
                brokers: localhost:9092
                configuration:
                  value.serde: JsonSerde
                  key.serde: StringSerde
                producer:
                  value.serde: JsonSerde
                  key.serde: StringSerde
                replication-factor: 1
  </code></pre>
  
  <p><strong>Versions</strong></p>
  
  <ul>
  <li>spring-boot: 2.2.4</li>
  <li>spring-cloud: Hoxton.SR1</li>
  <li>spring-cloud-stream-kafka-binder: 3.0.1</li>
  </ul>