How to fix 'Failed to update metadata after xxx ms' in spring-kafka when send record in callback

Question

spring-kafka can't send record in callback

    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, key, data);
    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
        @Override
        public void onFailure(Throwable ex) {
            log.error("log error...");
        }

        @Override
        public void onSuccess(SendResult<String, String> result) {
            kafkaTemplate.send("anotherTopic", "key", "data");
        }
    });

Kafka throw 'failed to update metadata' when I call kafkaTemplate.send() in onSuccess(),which is not expected

Answer 1

It looks like you can't perform producer operations on the callback thread - kafka-producer-network-thread - probably some deadlock in the producer code - waiting to get metadata which would use that same thread, so it times out.

You will probably need a second KafkaTemaplate (and producer factory because the default factory always returns the same producer).

Or simply execute the second send on a different thread...

@SpringBootApplication
public class So54492871Application {

    private static final ExecutorService exec = Executors.newSingleThreadExecutor();

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

    @Bean
    public NewTopic topic1() {
        return new NewTopic("so54492871-1", 1, (short) 1);
    }

    @Bean
    public NewTopic topic2() {
        return new NewTopic("so54492871-2", 1, (short) 1);
    }

    @Bean
    public ApplicationRunner runner(KafkaTemplate<String, String> template) {
        return args -> {
            ListenableFuture<SendResult<String, String>> future = template.send("so54492871-1", "foo");
            future.addCallback(result -> {
                System.out.println(Thread.currentThread().getName() + ":" + result);
                exec.execute(() -> {
                    ListenableFuture<SendResult<String, String>> future2 = template.send("so54492871-2", "bar");
                    future2.addCallback(result2 -> {
                        System.out.println(Thread.currentThread().getName() + ":" + result2);
                    }, ex -> {
                        System.out.println(ex.getMessage());
                    });
                });
            }, ex -> {
                System.out.println(ex.getMessage());
            });
            System.in.read();
        };
    }

}