Шаблон Kafka: исключение тайм-аута при отправке сообщения по шаблону kafka - PullRequest
Новая
       15

Шаблон Kafka: исключение тайм-аута при отправке сообщения по шаблону kafka

0 голосов
/ 04 января 2019

В весеннем загрузочном проекте с использованием шаблона 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') используются в качестве зависимости gradle вместе с spring-cloud-stream

1 Ответ

0 голосов
/ 06 января 2019

Только незначительные изменения были сделаны, чтобы заставить его работать 

@Service
@Slf4j
class KafkaMessagePublisherImpl {
   @Autowired
   private KafkaTemplate<String,String> kafkaAsyncPublisher;
   //Rest is same.

}
...