Kafka host lookup fails with kerberized zookeeper

[jvalkeal]

I was playing with Spring Cloud Dataflow Yarn with kerberized ambari install. Kafka and zookeeper are both using kerberos. When I ran TimeSourceKafkaApplication with this system, below error was raised. I believe ZookeeperConfiguration.doGetBrokerAddresses silently fails or someone is eating authentication exception if there is one.

java.lang.IllegalArgumentException: Host cannot be empty
    at org.springframework.util.Assert.hasText(Assert.java:168) ~[spring-core-4.2.6.RELEASE.jar!/:4.2.6.RELEASE]
    at org.springframework.integration.kafka.core.BrokerAddress.<init>(BrokerAddress.java:38) ~[spring-integration-kafka-1.3.1.RELEASE.jar!/:na]
    at org.springframework.integration.kafka.core.ZookeeperConfiguration$BrokerToBrokerAddressFunction.valueOf(ZookeeperConfiguration.java:101) ~[spring-integration-kafka-1.3.1.RELEASE.jar!/:na]
    at org.springframework.integration.kafka.core.ZookeeperConfiguration$BrokerToBrokerAddressFunction.valueOf(ZookeeperConfiguration.java:96) ~[spring-integration-kafka-1.3.1.RELEASE.jar!/:na]
    at com.gs.collections.impl.list.mutable.FastList.collect(FastList.java:943) ~[gs-collections-5.0.0.jar!/:na]
    at com.gs.collections.impl.list.mutable.FastList.collect(FastList.java:807) ~[gs-collections-5.0.0.jar!/:na]
    at org.springframework.integration.kafka.core.ZookeeperConfiguration.doGetBrokerAddresses(ZookeeperConfiguration.java:82) ~[spring-integration-kafka-1.3.1.RELEASE.jar!/:na]
    at org.springframework.integration.kafka.core.AbstractConfiguration.getBrokerAddresses(AbstractConfiguration.java:158) ~[spring-integration-kafka-1.3.1.RELEASE.jar!/:na]
    at org.springframework.integration.kafka.core.DefaultConnectionFactory.refreshMetadata(DefaultConnectionFactory.java:175) ~[spring-integration-kafka-1.3.1.RELEASE.jar!/:na]
    at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder$3.doWithRetry(KafkaMessageChannelBinder.java:409) ~[spring-cloud-stream-binder-kafka-1.0.2.RELEASE.jar!/:1.0.2.RELEASE]
    at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder$3.doWithRetry(KafkaMessageChannelBinder.java:405) ~[spring-cloud-stream-binder-kafka-1.0.2.RELEASE.jar!/:1.0.2.RELEASE]
    at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:263) ~[spring-retry-1.1.2.RELEASE.jar!/:na]
    at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:154) ~[spring-retry-1.1.2.RELEASE.jar!/:na]
    at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.ensureTopicCreated(KafkaMessageChannelBinder.java:405) [spring-cloud-stream-binder-kafka-1.0.2.RELEASE.jar!/:1.0.2.RELEASE]
    at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.doBindProducer(KafkaMessageChannelBinder.java:296) [spring-cloud-stream-binder-kafka-1.0.2.RELEASE.jar!/:1.0.2.RELEASE]
    at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.doBindProducer(KafkaMessageChannelBinder.java:121) [spring-cloud-stream-binder-kafka-1.0.2.RELEASE.jar!/:1.0.2.RELEASE]
    at org.springframework.cloud.stream.binder.AbstractBinder.bindProducer(AbstractBinder.java:184) [spring-cloud-stream-1.0.2.RELEASE.jar!/:1.0.2.RELEASE]
    at org.springframework.cloud.stream.binding.ChannelBindingService.bindProducer(ChannelBindingService.java:113) [spring-cloud-stream-1.0.2.RELEASE.jar!/:1.0.2.RELEASE]
    at org.springframework.cloud.stream.binding.BindableProxyFactory.bindOutputs(BindableProxyFactory.java:206) [spring-cloud-stream-1.0.2.RELEASE.jar!/:1.0.2.RELEASE]
    at org.springframework.cloud.stream.binding.OutputBindingLifecycle.start(OutputBindingLifecycle.java:57) [spring-cloud-stream-1.0.2.RELEASE.jar!/:1.0.2.RELEASE]
    at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:173) [spring-context-4.2.6.RELEASE.jar!/:4.2.6.RELEASE]
    at org.springframework.context.support.DefaultLifecycleProcessor.access$200(DefaultLifecycleProcessor.java:51) [spring-context-4.2.6.RELEASE.jar!/:4.2.6.RELEASE]
    at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.start(DefaultLifecycleProcessor.java:346) [spring-context-4.2.6.RELEASE.jar!/:4.2.6.RELEASE]
    at org.springframework.context.support.DefaultLifecycleProcessor.startBeans(DefaultLifecycleProcessor.java:149) [spring-context-4.2.6.RELEASE.jar!/:4.2.6.RELEASE]
    at org.springframework.context.support.DefaultLifecycleProcessor.onRefresh(DefaultLifecycleProcessor.java:112) [spring-context-4.2.6.RELEASE.jar!/:4.2.6.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:852) [spring-context-4.2.6.RELEASE.jar!/:4.2.6.RELEASE]
    at org.springframework.boot.context.embedded.EmbeddedWebApplicationContext.finishRefresh(EmbeddedWebApplicationContext.java:140) [spring-boot-1.3.5.RELEASE.jar!/:1.3.5.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:541) [spring-context-4.2.6.RELEASE.jar!/:4.2.6.RELEASE]
    at org.springframework.boot.context.embedded.EmbeddedWebApplicationContext.refresh(EmbeddedWebApplicationContext.java:118) [spring-boot-1.3.5.RELEASE.jar!/:1.3.5.RELEASE]
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:766) [spring-boot-1.3.5.RELEASE.jar!/:1.3.5.RELEASE]
    at org.springframework.boot.SpringApplication.createAndRefreshContext(SpringApplication.java:361) [spring-boot-1.3.5.RELEASE.jar!/:1.3.5.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:307) [spring-boot-1.3.5.RELEASE.jar!/:1.3.5.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1191) [spring-boot-1.3.5.RELEASE.jar!/:1.3.5.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1180) [spring-boot-1.3.5.RELEASE.jar!/:1.3.5.RELEASE]
    at org.springframework.cloud.stream.app.time.source.kafka.TimeSourceKafkaApplication.main(TimeSourceKafkaApplication.java:29) [time-source-kafka-1.0.0.BUILD-SNAPSHOT.jar!/:1.0.0.BUILD-SNAPSHOT]
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_91]
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_91]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_91]
    at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_91]
    at org.springframework.boot.loader.MainMethodRunner.run(MainMethodRunner.java:54) [time-source-kafka-1.0.0.BUILD-SNAPSHOT.jar!/:1.0.0.BUILD-SNAPSHOT]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_91]

Config I have which works ok with non-kerberized cluster is:

spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: sandbox.hortonworks.com:6667
          zkNodes: sandbox.hortonworks.com:2181

[jvalkeal]

Right, looks like master is not using I0Itec zkclient anymore which kinda changes whole story here. Was looking at ZkClient which uses system properties for sasl config which really sucks. I'd personally prefer to have a real boot props to handle all these.

[mbogoevici]

Thanks for reporting this Janne.

I think this is a story for the 0.9 client/binder. I'm wondering if we need to do anything special in Spring Kafka to address this (which we can pick up in Spring Cloud Stream via integration).

[artembilan]

@mbogoevici , @jvalkeal , just as a reminder. Do you still want anything to be done from Spring Integration Kafka side?
Thanks

[mbogoevici]

We can close this. There's nothing to be done here, especially with 1.3