title: ReplyingKafkaTemplate not getting response back
body: <p>Hello I am using ReplyingKafkaTemplate in my project.Configuration is as below</p>
  
  <pre><code>@Bean
      public KafkaMessageListenerContainer&lt;Object, Object&gt; replyListenerContainer() {
  
          ContainerProperties containerProperties = new ContainerProperties("account-get-response",
                  "account-put-response",
                  "account-post-response",
                  "account-delete-response");
          return new KafkaMessageListenerContainer&lt;&gt;(consumerFactory(), containerProperties);
      }
  
      @Bean
      public ReplyingKafkaTemplate&lt;Object, Object, Object&gt; replyKafkaTemplate(ProducerFactory&lt;Object, Object&gt; pf,
                                                                              KafkaMessageListenerContainer&lt;Object, Object&gt; lc) {
  
          ReplyingKafkaTemplate&lt;Object, Object, Object&gt; replyKafkaTemplate = new ReplyingKafkaTemplate&lt;&gt;(pf, lc);
          replyKafkaTemplate.setReplyTimeout(300000);
          replyKafkaTemplate.setSharedReplyTopic(true);
          return replyKafkaTemplate;
      }
  @Bean
      public ConsumerFactory&lt;Object, Object&gt; consumerFactory() {
  
          return new DefaultKafkaConsumerFactory&lt;&gt;(consumerConfigs());
      }
  
      @Bean
      public ProducerFactory&lt;Object, Object&gt; producerFactory() {
  
          return new DefaultKafkaProducerFactory&lt;&gt;(producerConfigs());
      }
  public Map&lt;String, Object&gt; consumerConfigs() {
  
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBootstrapServers);
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
          props.put(ConsumerConfig.GROUP_ID_CONFIG, "grp");
          props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
          props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
          return props;
      }
  
      public Map&lt;String, Object&gt; producerConfigs() {
  
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBootstrapServers);
          props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
          props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
          props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, kafkaRequestTimeoutMs);
          props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, kafkaDeliveryTimeoutMs);
          props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, kafkaRetryBackofMs);
          props.put(ProducerConfig.ACKS_CONFIG, kafkaackconfig);
          return props;
      }
  </code></pre>
  
  <p>The code is as below
  in this case requestTopic is account-post-request and responseTopic is account-post-response</p>
  
  <pre><code>ProducerRecord&lt;Object, Object&gt; record = new ProducerRecord&lt;Object, Object&gt;(requestTopic, jsonNode);
              record.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, responseTopic.getBytes()));
              record.headers().add(new RecordHeader("TransactionID", clientTxnId.getBytes()));
              LOGGER.info("TransactionID: " + clientTxnId + " Produced record on topic " + requestTopic);
              RequestReplyFuture&lt;Object, Object, Object&gt; sendAndReceive = replyKafkaTemplate.sendAndReceive(record);
              try {
                  LOGGER.info("TransactionID: " + clientTxnId + " Waiting for consumer response ");
                  reply = sendAndReceive.get().value();
                  LOGGER.info("TransactionID: " + clientTxnId + " Got the response from  consumer for topic :" + responseTopic);
              } catch (InterruptedException e) {
                  e.printStackTrace();
              } catch (ExecutionException e) {
                  e.printStackTrace();
              }
  </code></pre>
  
  <p>in this case consumer is consuming request the sending response back in response topic. but at producer end (replyingkafkatemplate)is it stuck at </p>
  
  <pre><code>reply = sendAndReceive.get().value();
  </code></pre>
  
  <p>means waiting for response
  But when I have checked in kafka tools response is getting written on desire topic on same correlation id.</p>
  
  <p><a href="https://i.stack.imgur.com/xBVb8.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/xBVb8.png" alt="enter image description here"></a></p>
  
  <p>I am not able to figure it out why it not able to read it from response topic even if it is available?
  Thanks in Advance.</p>