在spring引导项目中使用kafka模板发送消息

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

@Service
@Slf4j
class KafkaMessagePublisherImpl {
    @Autowired
    private KafkaTemplate kafkaAsyncPublisher;

    public void sendMessage() {
        ListenableFuture listenableFuture = kafkaAsyncPublisher.send(
                "test_topikc",
                "key_1",
                "{\"greeting\":\"Hello\"}");

        listenableFuture.addCallback(new ListenableFutureCallback<SendResult<?, ?>>() {

            @Override
            public void onSuccess(final SendResult<?, ?> message) {
                System.out.println("Sent");
            }

            @Override
            public void onFailure(final Throwable throwable) {

                System.out.println("Message sending failed");

            }
        });

    }

    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(
                ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "brokers1:9092,brokers2:9092");
        configProps.put(
                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class);
        configProps.put(
                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class);
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean(name = "kafkaAsyncPublisher")
    public KafkaTemplate<String, String> kafkaAsyncPublisher() {
        return new KafkaTemplate<>(producerFactory());
    }
}

在使用 sendMessage 方法，出现以下错误

Caused by: org.springframework.kafka.core.KafkaProducerException: Failed to send; nested exception is org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.
    at org.springframework.kafka.core.KafkaTemplate$1.onCompletion(KafkaTemplate.java:259)
    at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:760)
    at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:701)
    at org.springframework.kafka.core.DefaultKafkaProducerFactory$CloseSafeProducer.send(DefaultKafkaProducerFactory.java:170)
    at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:245)
    at org.springframework.kafka.core.KafkaTemplate.send(KafkaTemplate.java:157)
    at com.tesco.fps.messaging.service.impl.KafkaMessagePublisherImpl.lambda$sendMessage$0(KafkaMessagePublisherImpl.java:52)
    ... 34 common frames omitted
Caused by: org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.
``` `Springboot:1.5.10.RELEASE` 以及 `compile('org.apache.kafka:kafka_2.12:0.11.0.0')` 作为渐变依赖项与 `spring-cloud-stream`