Здравствуйте. Я использую ReplyingKafkaTemplate в своем проекте. Конфигурация указана ниже
@Bean
public KafkaMessageListenerContainer<Object, Object> replyListenerContainer() {
ContainerProperties containerProperties = new ContainerProperties("account-get-response",
"account-put-response",
"account-post-response",
"account-delete-response");
return new KafkaMessageListenerContainer<>(consumerFactory(), containerProperties);
}
@Bean
public ReplyingKafkaTemplate<Object, Object, Object> replyKafkaTemplate(ProducerFactory<Object, Object> pf,
KafkaMessageListenerContainer<Object, Object> lc) {
ReplyingKafkaTemplate<Object, Object, Object> replyKafkaTemplate = new ReplyingKafkaTemplate<>(pf, lc);
replyKafkaTemplate.setReplyTimeout(300000);
replyKafkaTemplate.setSharedReplyTopic(true);
return replyKafkaTemplate;
}
@Bean
public ConsumerFactory<Object, Object> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
@Bean
public ProducerFactory<Object, Object> producerFactory() {
return new DefaultKafkaProducerFactory<>(producerConfigs());
}
public Map<String, Object> consumerConfigs() {
Map<String, Object> props = new HashMap<>();
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<String, Object> producerConfigs() {
Map<String, Object> props = new HashMap<>();
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;
}
В этом случае код такой, как показано ниже. RequestTopi c - это учетная запись после запроса, а responseTopi c - учетная запись. -post-response
ProducerRecord<Object, Object> record = new ProducerRecord<Object, Object>(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<Object, Object, Object> 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();
}
в этом случае потребитель потребляет запрос на отправку ответа обратно в ответ topi c. но в конце производителя (replyingkafkatemplate) он застрял на
reply = sendAndReceive.get().value();
означает ожидание ответа Но когда я проверил в инструментах kafka, ответ записывается по желанию topi c с тем же идентификатором корреляции.
Я не могу понять, почему он не может прочитать его из ответа topi c, даже если он доступен? Заранее спасибо.