title: Sending object with Kafka / Spring Boot
body: <p>I just want to send my BackAccount object with producer, and I want to consume that object. I have Producer and Consumer Config, but it gives me error.</p>
  
  <p>My KafkaProducerConfig:</p>
  
  <pre><code>@Configuration
  public class KafkaProducerConfig {
  
  
      @Bean
      public Map&lt;String, Object&gt; producerConfigs() {
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
          props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
          props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
  
          return props;
      }
  
      @Bean
      public ProducerFactory&lt;String, BankAccount&gt; producerFactory() {
          return new DefaultKafkaProducerFactory&lt;&gt;(producerConfigs());
      }
  
      @Bean
      public KafkaTemplate&lt;String, BankAccount&gt; kafkaTemplate() {
          return new KafkaTemplate&lt;&gt;(producerFactory());
      }
  
  
  
  }
  </code></pre>
  
  <p>KafkaConsumerConfig : </p>
  
  <pre><code>@Configuration
  public class KafkaConsumerConfig {
  
  
      @Bean
      public Map&lt;String, Object&gt; consumerConfigs() {
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "");
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
          props.put(ConsumerConfig.GROUP_ID_CONFIG, "json");
          props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
          return props;
      }
  
      @Bean
      public ConsumerFactory&lt;String, BankAccount&gt; consumerFactory() {
          return new DefaultKafkaConsumerFactory&lt;&gt;(
                  consumerConfigs(),
                  new StringDeserializer(),
                  new JsonDeserializer&lt;&gt;(BankAccount.class));
      }
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, BankAccount&gt; kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;String, BankAccount&gt; factory =
                  new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactory());
          return factory;
      }
  
  }
  </code></pre>
  
  <p>It is really basic but my controller is :</p>
  
  <pre><code>public class KafkaController {
  
      @Autowired
      private KafkaTemplate&lt;String,BankAccount&gt; kafkaTemplate;
  
      private static final String TOPIC = "my-topic";
  
      @GetMapping("/publish/{message}")
      public String postKafka(@PathVariable("message") final String message){
          BankAccount bankAccount= new BankAccount();
          bankAccount.setBalance(120.0);
  
          kafkaTemplate.send(TOPIC,bankAccount);
          return "Succesfully";
      }
  
  
  }
  </code></pre>
  
  <p>I have error in KafkaConsumerConfig :</p>
  
  <pre><code>return new DefaultKafkaConsumerFactory&lt;&gt;(
                  consumerConfigs(),
                  new StringDeserializer(),
                  new JsonDeserializer&lt;&gt;(BankAccount.class));
  </code></pre>
  
  <p>in here : JsonDeserializer&lt;>(BankAccount.class)
  Diamond operator is not applicable for non-parametrized types</p>
  
  <p>I did the same thing in reference link: <a href="https://www.codenotfound.com/spring-kafka-json-serializer-deserializer-example.html" rel="nofollow noreferrer">https://www.codenotfound.com/spring-kafka-json-serializer-deserializer-example.html</a></p>
  
  <p>Can someone help me?</p>