如何修复spring kafka在回调中发送记录时“ ms后更新元数据失败”

zbdgwd5y  于 2021-06-06  发布在  Kafka
关注(0)|答案(1)|浏览(446)

spring kafka无法在回调中发送记录

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");
        }
    });

当我在onsuccess()中调用kafkatemplate.send()时,kafka抛出“未能更新元数据”,这是不应有的

8i9zcol2

8i9zcol21#

看起来你不能在回调线程上执行生产者操作- kafka-producer-network-thread -可能是生产者代码中的某个死锁—等待获取使用同一线程的元数据,因此超时。
你可能需要一点时间 KafkaTemaplate (和生产者工厂,因为默认工厂总是返回相同的生产者)。
或者在另一个线程上执行第二次发送。。。

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

}

相关问题