无待回复：消费者记录

Question

我正在尝试使用ReplyingKafkaTemplate，偶尔会看到下面的消息。

没有待处理的回复：消费者记录（主题=请求-回复-主题，分区=8，偏移量=1，创建时间=1544653843269，序列化密钥大小=-1，序列化值大小=1609，标题=记录标题（标题=[记录标题（键=kafka_correlationId，值=[-14,65,21，-118,70，-94,72,87，-113，-91,92,72，-124，-110，-64，-94]）]，isRead唯=假），键=null，带有相关ID：[-18271255759235816475365319231847350110]，可能超时，或使用共享回复主题

这将源于以下代码

RequestReplyFuture<K, V, R> 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));
  }
}

但只是间歇地发生

我还将共享回复主题设置为false，如下所示，并试图强制延长超时时间

ReplyingKafkaTemplate<String, Object, Object> replyKafkaTemplate = new ReplyingKafkaTemplate<>(pf, container);
        replyKafkaTemplate.setSharedReplyTopic(false);
        replyKafkaTemplate.setReplyTimeout(10000);
        return replyKafkaTemplate;

我的容器如下

@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, Object>> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());

    factory.setBatchListener(false);
    factory.getContainerProperties().setPollTimeout(1000);
    factory.getContainerProperties().setIdleEventInterval(10000L);
    factory.setConcurrency(3);
    factory.setReplyTemplate(kafkaTemplate());
    return factory;
}

Answer 1

已修复在消费者上使用属性@Header (KafkaHeaders)的问题。相关性_ID)

@KafkaListener(topics = "${kafka.topic.model}")
@SendTo("replymodeltopic")
@Override
public Model receive(ConsumerRecord<String, model> record, @Header(KafkaHeaders.CORRELATION_ID) byte[] correlation) { 
    record.headers().add(KafkaHeaders.CORRELATION_ID, correlation);
    return record.value();
}

在我的配置中

@Configuration
@EnableKafka
public class ReceiverConfig {

    @Value("${kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${kafka.tunnel.group}")
    private String tunnelGroup;

    @Value("${kafka.topic.json.reply}")
    private String jsonTopicReply;

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, tunnelGroup);

        return props;
    }

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        return props;
    }

    @Bean
    public ConsumerFactory<String, Model> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(), new JsonDeserializer<>(Model.class));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Model> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Model> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setReplyTemplate(replyTemplate());
        return factory;
    }

    @Bean
    public ProducerFactory<String, Model> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, Model> replyTemplate() {
        KafkaTemplate<String, Model> kafkaTemplate = new KafkaTemplate<>(producerFactory());
        kafkaTemplate.setDefaultTopic(jsonTopicReply);
        return kafkaTemplate;
    }

}

Answer 2

如果是间歇性的，则很可能回复花了很长时间才到达。信息似乎很清楚

可能超时，或者使用共享回复主题

每个客户端实例都必须使用自己的回复主题或专用分区。

编辑

如果收到的消息的相关 ID 与当前 this.futures 中的条目不匹配（挂起的答复），则会收到日志。这只能在以下情况下发生：

1. 请求超时（在这种情况下，将有相应的 WARN 日志）。
2. 模板是 stop（）ped（在这种情况下，this.futures 被清除）。
3. 由于某种原因（不应发生），已处理的回复会重新传递。
4. 在将密钥添加到 this.futures 之前收到回复（不会发生，因为它是在 send（） 处理记录之前插入的）。
5. 服务器端为同一请求发送 2 个或更多回复。
6. 其他一些应用程序正在向同一回复主题发送数据。如果您可以使用 DEBUG 日志记录重现它，这将有所帮助，因为这样我们也会在发送时记录相关密钥。