Не удалось создать привязку потребителя;повторная попытка через 30 секунд

Question

Я использую весеннюю облачную струю. У меня два канала. один - это использование kafka cluster1, другой - использование cluster2. конфиг как.

spring.cloud.stream.default-binder=kafka
spring.cloud.stream.binders.kafka.type=kafka
spring.cloud.stream.binders.kafka.environment.spring.cloud.stream.kafka.binder.brokers=xxxx

spring.cloud.stream.kafka.binder.auto-add-partitions=false
spring.cloud.stream.kafka.binder.auto-create-topics=false

spring.cloud.stream.bindings.channel1-input.destination=top1
spring.cloud.stream.bindings.channel1-input.group=mygroup
spring.cloud.stream.bindings.channel1-input.consumer.concurrency=3
spring.cloud.stream.bindings.channel1-input.consumer.partitioned=true
spring.cloud.stream.bindings.channel1-input.content-type=application/json


spring.cloud.stream.binders.kafka2.type=kafka
spring.cloud.stream.binders.kafka2.environment.spring.cloud.stream.kafka.binder.brokers=xxx

spring.cloud.stream.bindings.channel-input2.content-type=application/protocol-buffer
spring.cloud.stream.bindings.channel-input2.destination=topipc2
spring.cloud.stream.bindings.channel-input2.group=mygroup
spring.cloud.stream.bindings.channel-input2.producer.headerMode=raw
spring.cloud.stream.bindings.channel-input2.binder=kafka2

когда я запускаю приложение, я могу использовать сообщение из двух тем. но я получил ниже сообщение об ошибке. это меня смущает. у тебя есть идеи?

2019-04-17 19:51:54.502 [task-scheduler-4] ERROR o.s.cloud.stream.binding.BindingService - Failed to create consumer binding; retrying in 30 seconds
org.springframework.cloud.stream.binder.BinderException: Cannot initialize binder:
        at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.getPartitionsForTopic(KafkaTopicProvisioner.java:391)
        at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.getPartitionInfo(KafkaMessageChannelBinder.java:573)
        at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.createConsumerEndpoint(KafkaMessageChannelBinder.java:331)
        at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.createConsumerEndpoint(KafkaMessageChannelBinder.java:126)
        at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:279)
        at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:77)
        at org.springframework.cloud.stream.binder.AbstractBinder.bindConsumer(AbstractBinder.java:129)
        at org.springframework.cloud.stream.binding.BindingService.lambda$rescheduleConsumerBinding$0(BindingService.java:154)
        at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.NullPointerException: null
        at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.lambda$getPartitionsForTopic$2(KafkaTopicProvisioner.java:373)
        at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287)
        at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:164)
        at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.getPartitionsForTopic(KafkaTopicProvisioner.java:370)
        ... 15 common frames omitted

Answer 1

Наконец, я нашел основную причину, потому что я включил @EnableBinding (Sink.class), но мой сервер kafka не был автоматически создавать тему.после того, как я удалю @EnableBinding (Sink.class).проблема решена.

Answer 2

NPE является признаком ошибки на нашей стороне; Мы должны исправить это независимо, поэтому, пожалуйста, поднимите вопрос здесь .

Тем не менее, мне интересно, есть ли у вас проблемы с конфигурацией / безопасностью Kafka, которые мешают вам создавать разделы, так как кажется, что следующий код возвращает null

 CreatePartitionsResult partitions = adminClient
                        .createPartitions(Collections.singletonMap(topicName,
                                NewPartitions.increaseTo(effectivePartitionCount)));