ExecutionException:Due to: org.springframework.kafka.requestreply.KafkaReplyTimeoutException: Reply timed out using ReplyingKafkaTemplate

Question

I am using kafka to publish both async and sync messages to the broker .One listener would listen to the topic and respond for both sync and async calls. I am using same request topic for both the templates .. When using fire and forget(Async) I don't see any issues since listener would listen to the messages randomly from topic.When using synchronous call I am getting timeout exception.

Do I need to maintain multiple listeners for different templates ?
With same topic for both synchronous and async operations would there be any issues?

KafkaConfig.java

//Template for synchornous call

@Bean 
public ReplyingKafkaTemplate replyingKafkaTemplate (
    ProducerFactory pf,
    ConcurrentMessageListenerContainer repliesContainer)
{
    ReplyingKafkaTemplate replyTemplate =
        new ReplyingKafkaTemplate<>(pf, repliesContainer);
    replyTemplate.setSharedReplyTopic(true);
    return replyTemplate;
}

@Bean //register ConcurrentMessageListenerContainer bean
public ConcurrentMessageListenerContainer repliesContainer (
    ConcurrentKafkaListenerContainerFactory containerFactory)
{
    ConcurrentMessageListenerContainer repliesContainer =
        containerFactory.createContainer("responseTopic");
    repliesContainer.getContainerProperties().setGroupId(UUID.randomUUID().toString());
    repliesContainer.setAutoStartup(false);
    return repliesContainer;
}

//Template for asynchronous call

@Bean

@Qualifier("kafkaTemplate")
public KafkaTemplate kafkaTemplate (
    ProducerFactory pf,
    ConcurrentKafkaListenerContainerFactory factory)
{
    KafkaTemplate kafkaTemplate = new KafkaTemplate<>(pf);
    factory.setReplyTemplate(kafkaTemplate);
    return kafkaTemplate;
}

Here is service class

@Service
public class KafkaService
{
    @Autowired
    private ReplyingKafkaTemplate replyingKafkaTemplate;
    @Autowired
    private KafkaTemplate kafkaTemplate;
    @Autowired
    private KafkaConfig config;
    public Object sendAndReceive (Model model)
    {

        ProducerRecord producerRecord =
            new ProducerRecord("requestTopic", model);
        producerRecord.headers()
            .add(
                new RecordHeader(KafkaHeaders.REPLY_TOPIC, "replyTopic"));
        RequestReplyFuture replyFuture =
            replyingKafkaTemplate.sendAndReceive(producerRecord, Duration.ofSeconds(timeout));
        ConsumerRecord consumerRecord =
            replyFuture.get(timeout, TimeUnit.SECONDS);
        return consumerRecord.value();

    }
    public ResponseEntity

Gary Russell · Accepted Answer

This works exactly as I expected...

@SpringBootApplication
public class So73657031Application {

    public static void main(String[] args) {
        SpringApplication.run(So73657031Application.class, args);
    }

    @Bean
    ReplyingKafkaTemplate rkt(ProducerFactory pf,
            ConcurrentKafkaListenerContainerFactory factory,
            KafkaTemplate template) {

        factory.setReplyTemplate(template);
        ConcurrentMessageListenerContainer container = factory.createContainer("so73657031-replies");
        container.getContainerProperties().setGroupId("so73657031-replies");
        return new ReplyingKafkaTemplate<>(pf, container);
    }

    @Bean
    KafkaTemplate template(ProducerFactory pf) {
        return new KafkaTemplate<>(pf);
    }

    @Bean
    NewTopic topic1() {
        return TopicBuilder.name("so73657031").partitions(1).replicas(1).build();
    }

    @Bean
    NewTopic topic2() {
        return TopicBuilder.name("so73657031-replies").partitions(1).replicas(1).build();
    }

    @Bean
    public ApplicationRunner runner(ReplyingKafkaTemplate rTemplate,
            KafkaTemplate template) {

        return args -> {
            RequestReplyFuture future =
                    rTemplate.sendAndReceive(new ProducerRecord("so73657031", 0, null, "test"),
                            Duration.ofSeconds(30));
            System.out.println(future.getSendFuture().get(10, TimeUnit.SECONDS).getRecordMetadata());
            System.out.println(future.get(30, TimeUnit.SECONDS).value());
            ListenableFuture> future2 = template.send("so73657031", "oneWay");
            System.out.println(future2.get(10, TimeUnit.SECONDS).getRecordMetadata());
        };
    }

}

@Component
class Listener {


    @KafkaListener(id = "so73657031", topics = "so73657031")
    @SendTo
    String listen(String in) {
        System.out.println(in);
        return in.toUpperCase();
    }

}

logging.level.root=warn
logging.level.org.springframework.kafka.listener.adapter=debug

so73657031-0@2
2022-09-15 15:36:34.496 DEBUG 71184 --- [o73657031-0-C-1] .a.RecordMessagingMessageListenerAdapter : Processing [GenericMessage [payload=test, headers={kafka_offset=2, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@1582e8e4, kafka_correlationId=[B@2a266829, kafka_timestampType=CREATE_TIME, kafka_deliveryAttempt=1, kafka_replyTopic=[B@3dad3e81, kafka_receivedPartitionId=0, kafka_receivedTopic=so73657031, kafka_receivedTimestamp=1663270594381, kafka_groupId=so73657031}]]
test
2022-09-15 15:36:34.499 DEBUG 71184 --- [o73657031-0-C-1] .a.RecordMessagingMessageListenerAdapter : Listener method returned result [TEST] - generating response message for it
TEST
so73657031-0@3
2022-09-15 15:36:34.519 DEBUG 71184 --- [o73657031-0-C-1] .a.RecordMessagingMessageListenerAdapter : Processing [GenericMessage [payload=oneWay, headers={kafka_offset=3, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@1582e8e4, kafka_timestampType=CREATE_TIME, kafka_deliveryAttempt=1, kafka_receivedPartitionId=0, kafka_receivedTopic=so73657031, kafka_receivedTimestamp=1663270594514, kafka_groupId=so73657031}]]
oneWay
2022-09-15 15:36:34.519 DEBUG 71184 --- [o73657031-0-C-1] .a.RecordMessagingMessageListenerAdapter : Listener method returned result [ONEWAY] - generating response message for it
2022-09-15 15:36:34.519 DEBUG 71184 --- [o73657031-0-C-1] .a.RecordMessagingMessageListenerAdapter : No replyTopic to handle the reply: ONEWAY

ExecutionException:Due to: org.springframework.kafka.requestreply.KafkaReplyTimeoutException: Reply timed out using ReplyingKafkaTemplate

//Template for synchornous call

//Template for asynchronous call

Here is service class

Here is the listener class.

application.properties

Answers (1)

Related Questions