title: IllegalStateException: please consider configuring an 'ErrorHandlingDeserializer' in the value and/or key deserializer
body: <p>I want to implement Kafka producer which sends and receives Java Serialized Objects. I tried this:</p>
  <p><em><strong>Producer:</strong></em></p>
  <pre><code>@Configuration
  public class KafkaProducerConfig {
  
  @Value(value = &quot;${kafka.bootstrapAddress}&quot;)
  private String bootstrapAddress;
  
  @Bean
  public ProducerFactory&lt;String, SaleRequestFactory&gt; saleRequestFactoryProducerFactory() {
      Map&lt;String, Object&gt; configProps = new HashMap&lt;&gt;();
      configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
      configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
      configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, SaleRequestFactorySerializer.class);
      return new DefaultKafkaProducerFactory&lt;&gt;(configProps);
  }
  
  @Bean
  public ProducerFactory&lt;String, String&gt; producerFactory() {
      Map&lt;String, Object&gt; configProps = new HashMap&lt;&gt;();
      configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
      return new DefaultKafkaProducerFactory&lt;&gt;(configProps);
  }
  
  
  @Bean
  public KafkaTemplate&lt;String, SaleRequestFactory&gt; saleRequestFactoryKafkaTemplate() {
      return new KafkaTemplate&lt;&gt;(saleRequestFactoryProducerFactory());
  }
  
  @Bean
  public KafkaTemplate&lt;String, String&gt; kafkaTemplate() {
      return new KafkaTemplate&lt;&gt;(producerFactory());
  }
  </code></pre>
  <p>}</p>
  <p><em><strong>Send object:</strong></em></p>
  <pre><code>@Autowired
  private KafkaTemplate&lt;String, SaleRequestFactory&gt; saleRequestFactoryKafkaTemplate;
  
  private static String topic = &quot;tp-sale&quot;;
  
  private void perform(){
  
      SaleRequestFactory obj = new SaleRequestFactory();
      obj.setId(100);
  
      ListenableFuture&lt;SendResult&lt;String, SaleRequestFactory&gt;&gt; send = saleRequestFactoryKafkaTemplate.send(topic, obj);
  }
  </code></pre>
  <p><em><strong>Consumer:</strong></em></p>
  <pre><code>@EnableKafka
  @Configuration
  public class KafkaConsumerConfig {
  
      @Value(value = &quot;${kafka.bootstrapAddress}&quot;)
      private String bootstrapAddress;
  
      private String groupId = &quot;test&quot;;
  
      @Bean
      public ConsumerFactory&lt;String, SaleResponseFactory&gt; consumerFactory() {
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
          props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, SaleResponseFactoryDeserializer.class);
          return new DefaultKafkaConsumerFactory&lt;&gt;(props);
      }
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, SaleResponseFactory&gt; kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;String, SaleResponseFactory&gt; factory =
                  new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactory());
          return factory;
      }
  }
  </code></pre>
  <p><em><strong>Receive Object</strong></em></p>
  <pre><code>    @KafkaListener(topics = &quot;tp-sale&quot;)
  public SaleResponseFactory transactionElavonAuthorizeProcess(@Payload SaleRequestFactory tf, @Headers MessageHeaders headers) throws Exception {
  
      System.out.println(tf.getId());
  
      SaleResponseFactory resObj = new SaleResponseFactory();
      resObj.setUnique_id(&quot;123123&quot;);
  
      return resObj;
  }
  </code></pre>
  <p><em><strong>Custom objects</strong></em></p>
  <pre><code>    @Getter
      @Setter
      @NoArgsConstructor
      @AllArgsConstructor
      @Builder(toBuilder = true)
      public class SaleRequestFactory implements Serializable{
      
          private static final long serialVersionUID = 1744050117179344127L;
          
          private int id;
      }
  
  public class SaleRequestFactorySerializer implements Serializable, Serializer&lt;SaleRequestFactory&gt; {
  
      @Override
      public byte[] serialize(String topic, SaleRequestFactory data) {
          // convert data to byte[]
          ByteArrayOutputStream out = new ByteArrayOutputStream();
          try
          {
              ObjectOutputStream outputStream = new ObjectOutputStream(out);
              outputStream.writeObject(data);
              out.close();
          }
          catch (IOException e)
          {
              e.printStackTrace();
          }
  
          return out.toByteArray();
      }
  }
  
  
      @Getter
      @Setter
      @NoArgsConstructor
      @AllArgsConstructor
      @Builder(toBuilder = true)
      public class SaleResponseFactory implements Serializable{
      
          private static final long serialVersionUID = 1744050117179344127L;
      
          private String unique_id;
      }
  
  public class SaleResponseFactoryDeserializer implements Serializable, Deserializer&lt;SaleResponseFactory&gt; {
  
      @Override
      public SaleResponseFactory deserialize(String topic, byte[] data) {
          // convert data to SaleResponseFactory
          SaleResponseFactory saleResponseFactory = null;
          try
          {
              ByteArrayInputStream bis = new ByteArrayInputStream(data);
              ObjectInputStream in = new ObjectInputStream(bis);
              saleResponseFactory = (SaleResponseFactory) in.readObject();
              in.close();
          }
          catch (IOException | ClassNotFoundException e)
          {
              e.printStackTrace();
          }
          return saleResponseFactory;
      }
  }
  </code></pre>
  <p>When I try to send message I get error:</p>
  <pre><code>Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition tp-sale-0 at offset 0. If needed, please seek past the record to continue consumption.
  Caused by: java.lang.ClassCastException: null
  21:27:51.152 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] DEBUG KafkaMessageListenerContainer$ListenerConsumer[debug:296] - Commit list: {}
  21:27:51.153 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] ERROR KafkaMessageListenerContainer$ListenerConsumer[error:149] - Consumer exception
  java.lang.IllegalStateException: This error handler cannot process 'SerializationException's directly; please consider configuring an 'ErrorHandlingDeserializer' in the value and/or key deserializer
          at org.springframework.kafka.listener.SeekUtils.seekOrRecover(SeekUtils.java:145)
          at org.springframework.kafka.listener.SeekToCurrentErrorHandler.handle(SeekToCurrentErrorHandler.java:103)
          at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.handleConsumerException(KafkaMessageListenerContainer.java:1265)
          at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1022)
          at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
          at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
          at java.base/java.lang.Thread.run(Thread.java:835)
  Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition tp-sale-0 at offset 0. If needed, please seek past the record to continue consumption.
  Caused by: java.lang.ClassCastException: null
  </code></pre>
  <p>Do you know how I can fix this issue?</p>
  <p><em><strong>EDIT:</strong></em>
  I managed to implement these improvements:</p>
  <p><em><strong>Producer:</strong></em></p>
  <pre><code>@Configuration
  public class KafkaProducerConfig {
  
  @Value(value = &quot;${kafka.bootstrapAddress}&quot;)
  private String bootstrapAddress;
  
  @Bean
  public ProducerFactory&lt;String, SaleRequestFactory&gt; saleRequestFactoryProducerFactory() {
      Map&lt;String, Object&gt; configProps = new HashMap&lt;&gt;();
      configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
      configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
      configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, SaleRequestFactorySerializer.class);
      return new DefaultKafkaProducerFactory&lt;&gt;(configProps);
  }
  
  @Bean
  public ProducerFactory&lt;String, String&gt; producerFactory() {
      Map&lt;String, Object&gt; configProps = new HashMap&lt;&gt;();
      configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
      return new DefaultKafkaProducerFactory&lt;&gt;(configProps);
  }
  
  
  @Bean
  public KafkaTemplate&lt;String, SaleRequestFactory&gt; saleRequestFactoryKafkaTemplate() {
      return new KafkaTemplate&lt;&gt;(saleRequestFactoryProducerFactory());
  }
  
  @Bean
  public KafkaTemplate&lt;String, String&gt; kafkaTemplate() {
      return new KafkaTemplate&lt;&gt;(producerFactory());
  }
  </code></pre>
  <p>}</p>
  <p><em><strong>Send object:</strong></em></p>
  <pre><code>@Autowired
  private KafkaTemplate&lt;String, SaleRequestFactory&gt; saleRequestFactoryKafkaTemplate;
  
  private static String topic = &quot;tp-sale&quot;;
  
  private void perform(){
  
      SaleRequestFactory obj = new SaleRequestFactory();
      obj.setId(100);
  
      ListenableFuture&lt;SendResult&lt;String, SaleRequestFactory&gt;&gt; send = saleRequestFactoryKafkaTemplate.send(topic, obj);
  }
  </code></pre>
  <p><em><strong>Consumer:</strong></em></p>
  <pre><code>@EnableKafka
  @Configuration
  public class KafkaConsumerConfig {
  
      @Value(value = &quot;${kafka.bootstrapAddress}&quot;)
      private String bootstrapAddress;
  
      private String groupId = &quot;test&quot;;
  
      @Bean
      public ConsumerFactory&lt;String, SaleResponseFactory&gt; consumerFactory() {
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
          props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, SaleResponseFactoryDeserializer.class);
          return new DefaultKafkaConsumerFactory&lt;&gt;(props);
      }
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, SaleResponseFactory&gt; kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;String, SaleResponseFactory&gt; factory =
                  new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactory());
          return factory;
      }
  }
  </code></pre>
  <p><em><strong>Receive Object</strong></em></p>
  <pre><code>    @KafkaListener(topics = &quot;tp-sale&quot;)
  public SaleResponseFactory transactionElavonAuthorizeProcess(@Payload SaleRequestFactory tf, @Headers MessageHeaders headers) throws Exception {
  
      System.out.println(tf.getId());
  
      SaleResponseFactory resObj = new SaleResponseFactory();
      resObj.setUnique_id(&quot;123123&quot;);
  
      return resObj;
  }
  </code></pre>
  <p><em><strong>Custom objects</strong></em></p>
  <pre><code>    @Getter
      @Setter
      @NoArgsConstructor
      @AllArgsConstructor
      @Builder(toBuilder = true)
      public class SaleRequestFactory implements Serializable{
      
          private static final long serialVersionUID = 1744050117179344127L;
          
          private int id;
      }
  
  public class SaleRequestFactorySerializer implements Serializable, Serializer&lt;SaleRequestFactory&gt; {
  
      @Override
      public byte[] serialize(String topic, SaleRequestFactory data) {
          // convert data to byte[]
          ByteArrayOutputStream out = new ByteArrayOutputStream();
          try
          {
              ObjectOutputStream outputStream = new ObjectOutputStream(out);
              outputStream.writeObject(data);
              out.close();
          }
          catch (IOException e)
          {
              e.printStackTrace();
          }
  
          return out.toByteArray();
      }
  }
  
  
      @Getter
      @Setter
      @NoArgsConstructor
      @AllArgsConstructor
      @Builder(toBuilder = true)
      public class SaleResponseFactory implements Serializable{
      
          private static final long serialVersionUID = 1744050117179344127L;
      
          private String unique_id;
      }
  
  public class SaleResponseFactoryDeserializer implements Serializable, Deserializer&lt;SaleResponseFactory&gt; {
  
      @Override
      public SaleResponseFactory deserialize(String topic, byte[] data) {
          // convert data to SaleResponseFactory
          SaleResponseFactory saleResponseFactory = null;
          try
          {
              ByteArrayInputStream bis = new ByteArrayInputStream(data);
              ObjectInputStream in = new ObjectInputStream(bis);
              saleResponseFactory = (SaleResponseFactory) in.readObject();
              in.close();
          }
          catch (IOException | ClassNotFoundException e)
          {
              e.printStackTrace();
          }
          return saleResponseFactory;
      }
  }
  </code></pre>
  <p>When I send some message I get error:</p>
  <pre><code>13:03:53.675 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] DEBUG RecordMessagingMessageListenerAdapter[debug:296] - Listener method returned result [org.factory.SaleResponseFactory@69c400ab] - generating response message for it
  13:03:53.675 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] DEBUG RecordMessagingMessageListenerAdapter[debug:296] - No replyTopic to handle the reply: org.factory.SaleResponseFactory@69c400ab
  </code></pre>
  <p>Do you know how I can return <code>SaleResponseFactory</code> back to the original Producer class?</p>
  <p>EDIT 2:
  I get this exception:</p>
  <pre><code>org.springframework.beans.factory.NoSuchBeanDefinitionException: No qualifying bean of type 'org.springframework.kafka.listener.KafkaMessageListenerContainer&lt;java.lang.String, org.engine.plugin.transactions.factory.SaleResponseFactory&gt;' available: expected at least 1 bean which qualifies as autowire candidate. Dependency annotations: {}
          at org.springframework.beans.factory.support.DefaultListableBeanFactory.raiseNoMatchingBeanFound(DefaultListableBeanFactory.java:1717)
  </code></pre>