title: ClassCastException: class java.lang.String cannot be cast to class SaleResponseFactory
body: <p>I want to implement Kafka Consumer and Producer which sends and receives Java Objects. Full <a href="https://github.com/rcbandit111/kafka_test" rel="nofollow noreferrer">Source</a> 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());
      }
  
      @Bean
      public ReplyingKafkaTemplate&lt;String, SaleRequestFactory, SaleResponseFactory&gt; replyKafkaTemplate(ProducerFactory&lt;String, SaleRequestFactory&gt; producerFactory, ConcurrentKafkaListenerContainerFactory&lt;String, SaleResponseFactory&gt; factory) {
          ConcurrentMessageListenerContainer&lt;String, SaleResponseFactory&gt; kafkaMessageListenerContainer = factory.createContainer(&quot;tp-sale&quot;);
          kafkaMessageListenerContainer.getContainerProperties().setGroupId(&quot;tp-sale.reply&quot;);
          return new ReplyingKafkaTemplate&lt;&gt;(producerFactory, kafkaMessageListenerContainer);
      }
  }
  </code></pre>
  <p><em><strong>Send object:</strong></em></p>
  <pre><code> @RestController
  @RequestMapping(&quot;/checkout&quot;)
  public class CheckoutController {
      
      private TransactionService transactionService;
      private KafkaTemplate&lt;String, SaleRequestFactory&gt; saleRequestFactoryKafkaTemplate;
      private ReplyingKafkaTemplate&lt;String, SaleRequestFactory, SaleResponseFactory&gt; requestReplyKafkaTemplate;
      private static String topic = &quot;tp-sale&quot;;
  
      @Autowired
      public CheckoutController(ValidationMessage validationMessage, TransactionService transactionService,
                                KafkaTemplate&lt;String, SaleRequestFactory&gt; saleRequestFactoryKafkaTemplate,
                                ReplyingKafkaTemplate&lt;String, SaleRequestFactory, SaleResponseFactory&gt; requestReplyKafkaTemplate){
          this.transactionService = transactionService;
          this.saleRequestFactoryKafkaTemplate = saleRequestFactoryKafkaTemplate;
          this.requestReplyKafkaTemplate = requestReplyKafkaTemplate;
      }
  
      @PostMapping(&quot;test&quot;)
      private void performPayment() throws ExecutionException, InterruptedException, TimeoutException {
  
          Transaction transaction = new Transaction();
          transaction.setStatus(PaymentTransactionStatus.IN_PROGRESS.getText());
  
          Transaction insertedTransaction = transactionService.save(transaction);
  
          SaleRequestFactory obj = new SaleRequestFactory();
          obj.setId(100);
  
          ProducerRecord&lt;String, SaleRequestFactory&gt; record = new ProducerRecord&lt;&gt;(&quot;tp-sale&quot;, obj);
          RequestReplyFuture&lt;String, SaleRequestFactory, SaleResponseFactory&gt; replyFuture = requestReplyKafkaTemplate.sendAndReceive(record);
          SendResult&lt;String, SaleRequestFactory&gt; sendResult = replyFuture.getSendFuture().get(10, TimeUnit.SECONDS);
          ConsumerRecord&lt;String, SaleResponseFactory&gt; consumerRecord = replyFuture.get(10, TimeUnit.SECONDS);
  
  
          SaleResponseFactory value = consumerRecord.value();
          System.out.println(&quot;!!!!!!!!!!!! &quot; + value.getUnique_id());
      }
  }
  </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>@Component
  public class ProcessingSaleListener {
  
      private static String topic = &quot;tp-sale&quot;;
  
      @KafkaListener(topics = &quot;tp-sale&quot;)
      public SaleResponseFactory process(@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>import java.io.Serializable;
  
  import lombok.AllArgsConstructor;
  import lombok.Builder;
  import lombok.Getter;
  import lombok.NoArgsConstructor;
  import lombok.Setter;
  
  @Getter
  @Setter
  @NoArgsConstructor
  @AllArgsConstructor
  @Builder(toBuilder = true)
  public class SaleRequestFactory implements Serializable {
  
      private static final long serialVersionUID = 1744050117179344127L;
      
      private int id;
  
  }
  </code></pre>
  <p><em><strong>Serializer</strong></em></p>
  <pre><code>import org.apache.kafka.common.serialization.Serializer;
  
  import java.io.ByteArrayOutputStream;
  import java.io.IOException;
  import java.io.ObjectOutputStream;
  import java.io.Serializable;
  
  public class SaleRequestFactorySerializer implements Serializable, Serializer&lt;SaleRequestFactory&gt; {
  
      @Override
      public byte[] serialize(String topic, SaleRequestFactory data)
      {
          ByteArrayOutputStream out = new ByteArrayOutputStream();
          try
          {
              ObjectOutputStream outputStream = new ObjectOutputStream(out);
              outputStream.writeObject(data);
              out.close();
          }
          catch (IOException e)
          {
              throw new RuntimeException(&quot;Unhandled&quot;, e);
          }
          return out.toByteArray();
      }
  }
  </code></pre>
  <p><em><strong>Response Object</strong></em></p>
  <pre><code>import java.io.Serializable;
  import java.time.LocalDateTime;
  
  import lombok.AllArgsConstructor;
  import lombok.Builder;
  import lombok.Getter;
  import lombok.NoArgsConstructor;
  import lombok.Setter;
  
  @Getter
  @Setter
  @NoArgsConstructor
  @AllArgsConstructor
  @Builder(toBuilder = true)
  public class SaleResponseFactory implements Serializable {
  
      private static final long serialVersionUID = 1744050117179344127L;
  
      private String unique_id;
  }
  </code></pre>
  <p><em><strong>Response Class</strong></em></p>
  <pre><code>import org.apache.kafka.common.serialization.Deserializer;
  
  import java.io.ByteArrayInputStream;
  import java.io.IOException;
  import java.io.ObjectInputStream;
  import java.io.Serializable;
  
  public class SaleResponseFactoryDeserializer implements Serializable, Deserializer&lt;SaleRequestFactory&gt; {
  
      @Override
      public SaleRequestFactory deserialize(String topic, byte[] data)
      {
          SaleRequestFactory saleRequestFactory = null;
          try
          {
              ByteArrayInputStream bis = new ByteArrayInputStream(data);
              ObjectInputStream in = new ObjectInputStream(bis);
              saleRequestFactory = (SaleRequestFactory) in.readObject();
              in.close();
          }
          catch (IOException | ClassNotFoundException e)
          {
              throw new RuntimeException(&quot;Unhandled&quot;, e);
          }
          return saleRequestFactory;
      }
  }
  </code></pre>
  <p>When I send some message I get error:</p>
  <pre><code>23:52:08.920 [consumer-0-C-1] DEBUG KafkaMessageListenerContainer$ListenerConsumer[debug:313] - Commit list: {tp-sale-0=OffsetAndMetadata{offset=35, leaderEpoch=null, metadata=''}}
  23:52:08.920 [consumer-0-C-1] DEBUG KafkaMessageListenerContainer$ListenerConsumer[debug:313] - Committing: {tp-sale-0=OffsetAndMetadata{offset=35, leaderEpoch=null, metadata=''}}
  23:52:08.926 [http-nio-8090-exec-4] DEBUG OpenEntityManagerInViewInterceptor[afterCompletion:111] - Closing JPA EntityManager in OpenEntityManagerInViewInterceptor
  23:52:08.928 [http-nio-8090-exec-4] DEBUG DispatcherServlet[logResult:1101] - Failed to complete request: java.lang.ClassCastException: class java.lang.String cannot be cast to class org.engine.plugin.transactions.factory.SaleResponseFactory (java.lang.String is in module java.base of loader 'bootstrap'; org.engine.plugin.transactions.factory.SaleResponseFactory is in unnamed module of loader org.springframework.boot.loader.LaunchedURLClassLoader @6267c3bb)
  23:52:08.928 [http-nio-8090-exec-4] DEBUG HstsHeaderWriter[writeHeaders:169] - Not injecting HSTS header since it did not match the requestMatcher org.springframework.security.web.header.writers.HstsHeaderWriter$SecureRequestMatcher@2cfd2bb6
  23:52:08.928 [http-nio-8090-exec-4] DEBUG HttpSessionSecurityContextRepository[saveContext:351] - SecurityContext is empty or contents are anonymous - context will not be stored in HttpSession.
  23:52:08.928 [http-nio-8090-exec-4] DEBUG SecurityContextPersistenceFilter[doFilter:119] - SecurityContextHolder now cleared, as request processing completed
  23:52:08.929 [http-nio-8090-exec-4] ERROR [dispatcherServlet][log:175] - Servlet.service() for servlet [dispatcherServlet] in context with path [/engine] threw exception [Request processing failed; nested exception is java.lang.ClassCastException: class java.lang.String cannot be cast to class org.engine.plugin.transactions.factory.SaleResponseFactory (java.lang.String is in module java.base of loader 'bootstrap'; org.engine.plugin.transactions.factory.SaleResponseFactory is in unnamed module of loader org.springframework.boot.loader.LaunchedURLClassLoader @6267c3bb)] with root cause
  java.lang.ClassCastException: class java.lang.String cannot be cast to class org.engine.plugin.transactions.factory.SaleResponseFactory (java.lang.String is in module java.base of loader 'bootstrap'; org.engine.plugin.transactions.factory.SaleResponseFactory is in unnamed module of loader org.springframework.boot.loader.LaunchedURLClassLoader @6267c3bb)
          at org.engine.rest.CheckoutController.performPayment(CheckoutController.java:95)
          at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
          at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
          at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
          at java.base/java.lang.reflect.Method.invoke(Method.java:567)
          at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:190)
          at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:138)
          at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:105)
          at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:878)
          at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:792)
          at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:87)
          at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:1040)
          at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:943)
          at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:1006)
          at org.springframework.web.servlet.FrameworkServlet.doPost(FrameworkServlet.java:909)
          at javax.servlet.http.HttpServlet.service(HttpServlet.java:652)
          at org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:883)
          at javax.servlet.http.HttpServlet.service(HttpServlet.java:733)
          at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:231)
          at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
          at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:53)
          at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
          at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
          at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:320)
          at org.springframework.security.web.access.intercept.FilterSecurityInterceptor.invoke(FilterSecurityInterceptor.java:126)
          at org.springframework.security.web.access.intercept.FilterSecurityInterceptor.doFilter(FilterSecurityInterceptor.java:90)
          at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:334)
          at org.springframework.security.web.access.ExceptionTranslationFilter.doFilter(ExceptionTranslationFilter.java:118)
          at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:334)
          at org.springframework.security.web.session.SessionManagementFilter.doFilter(SessionManagementFilter.java:137)
          at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:334)
          at org.springframework.security.web.authentication.AnonymousAuthenticationFilter.doFilter(AnonymousAuthenticationFilter.java:111)
          at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:334)
          at org.springframework.security.web.servletapi.SecurityContextHolderAwareRequestFilter.doFilter(SecurityContextHolderAwareRequestFilter.java:158)
          at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:334)
          at org.springframework.security.web.savedrequest.RequestCacheAwareFilter.doFilter(RequestCacheAwareFilter.java:63)
          at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:334)
          at org.springframework.security.web.authentication.logout.LogoutFilter.doFilter(LogoutFilter.java:116)
          at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:334)
          at org.springframework.security.web.header.HeaderWriterFilter.doHeadersAfter(HeaderWriterFilter.java:92)
          at org.springframework.security.web.header.HeaderWriterFilter.doFilterInternal(HeaderWriterFilter.java:77)
          at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)
          at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:334)
          at org.springframework.security.web.context.SecurityContextPersistenceFilter.doFilter(SecurityContextPersistenceFilter.java:105)
          at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:334)
          at org.springframework.security.web.context.request.async.WebAsyncManagerIntegrationFilter.doFilterInternal(WebAsyncManagerIntegrationFilter.java:56)
          at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)
          at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:334)
          at org.springframework.security.web.FilterChainProxy.doFilterInternal(FilterChainProxy.java:215)
          at org.springframework.security.web.FilterChainProxy.doFilter(FilterChainProxy.java:178)
          at org.springframework.web.filter.DelegatingFilterProxy.invokeDelegate(DelegatingFilterProxy.java:358)
          at org.springframework.web.filter.DelegatingFilterProxy.doFilter(DelegatingFilterProxy.java:271)
          at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
          at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
          at org.springframework.web.filter.RequestContextFilter.doFilterInternal(RequestContextFilter.java:100)
          at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)
          at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
          at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
          at org.springframework.web.filter.FormContentFilter.doFilterInternal(FormContentFilter.java:93)
          at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)
          at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
          at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
          at org.springframework.boot.actuate.metrics.web.servlet.WebMvcMetricsFilter.doFilterInternal(WebMvcMetricsFilter.java:93)
          at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)
          at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
          at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
          at org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:201)
          at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)
          at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
          at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
          at org.springframework.web.filter.CorsFilter.doFilterInternal(CorsFilter.java:92)
          at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)
          at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
          at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
          at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:202)
          at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:97)
          at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:542)
          at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:143)
          at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:92)
          at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:78)
          at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:343)
          at org.apache.coyote.http11.Http11Processor.service(Http11Processor.java:374)
          at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:65)
          at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:868)
          at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1590)
          at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:49)
          at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
          at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
          at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)
          at java.base/java.lang.Thread.run(Thread.java:835)
  </code></pre>
  <p>Do you know how I can fix this issue?</p>