ReplyingKafkaTemplate not getting response back

Hello I am using ReplyingKafkaTemplate in my project.Configuration is as below

@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;
    }

The code is as below in this case requestTopic is account-post-request and responseTopic is account-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();
            }

in this case consumer is consuming request the sending response back in response topic. but at producer end (replyingkafkatemplate)is it stuck at

reply = sendAndReceive.get().value();

means waiting for response But when I have checked in kafka tools response is getting written on desire topic on same correlation id.

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.

Solution

It's probably a race condition; set ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to earliest.

It defaults to latest; if the response arrives before the consumer is subscribed to the response topic, it won't get it.