title: Failed to create consumer binding; retrying in 30 seconds
body: <p>I am using spring cloud stream. I have two channel. one is use kafka cluster1, one is use cluster2. config is like.</p>
  
  <pre><code>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
  </code></pre>
  
  <p>when I run application, I am able consume two topic's message. but I got below error message. that make me confusing. do you have any ideas?</p>
  
  <pre><code>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
  
  </code></pre>