title: No pending reply: ConsumerRecord
body: <p>I am trying to use ReplyingKafkaTemplate, and intermittently I keep seeing the message below.</p>
  
  <blockquote>
    <p>No pending reply: ConsumerRecord(topic = request-reply-topic,
    partition = 8, offset = 1, CreateTime = 1544653843269, serialized key
    size = -1, serialized value size = 1609, headers =
    RecordHeaders(headers = [RecordHeader(key = kafka_correlationId, value
    = [-14, 65, 21, -118, 70, -94, 72, 87, -113, -91, 92, 72, -124, -110, -64, -94])], isReadOnly = false), key = null, with correlationId: [-18271255759235816475365319231847350110], perhaps timed out, or using
    a shared reply topic</p>
  </blockquote>
  
  <p>It would stem from the code below</p>
  
  <pre><code>RequestReplyFuture&lt;K, V, R&gt; future = this.futures.remove(correlationId);
  if (future == null) {
    if (this.sharedReplyTopic) {
      if (this.logger.isDebugEnabled()) {
        this.logger.debug(missingCorrelationLogMessage(record, correlationId));
      }
    }
    else if (this.logger.isErrorEnabled()) {
      this.logger.error(missingCorrelationLogMessage(record, correlationId));
    }
  }
  </code></pre>
  
  <p>But happens only intemittently</p>
  
  <p>I have also set the shared replyTopic to false as below and attempted to force a longer timeout</p>
  
  <pre><code>ReplyingKafkaTemplate&lt;String, Object, Object&gt; replyKafkaTemplate = new ReplyingKafkaTemplate&lt;&gt;(pf, container);
          replyKafkaTemplate.setSharedReplyTopic(false);
          replyKafkaTemplate.setReplyTimeout(10000);
          return replyKafkaTemplate;
  </code></pre>
  
  <p>My Container is as below</p>
  
  <pre><code>@Bean
  public KafkaListenerContainerFactory&lt;ConcurrentMessageListenerContainer&lt;String, Object&gt;&gt; kafkaListenerContainerFactory() {
      ConcurrentKafkaListenerContainerFactory&lt;String, Object&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
      factory.setConsumerFactory(consumerFactory());
  
      factory.setBatchListener(false);
      factory.getContainerProperties().setPollTimeout(1000);
      factory.getContainerProperties().setIdleEventInterval(10000L);
      factory.setConcurrency(3);
      factory.setReplyTemplate(kafkaTemplate());
      return factory;
  }
  </code></pre>