Getting Bootstrap broker ip:9092 disconnected error from kafka spout

Versions:

"org.apache.storm" % "storm-kafka-client" % "1.2.1"
"org.apache.storm" % "storm-core" % "1.2.1" % "compile"
Kafka: 0.10.1.0


I am getting following error/warnings, running in localCluster, from my kafka spout:

2018-06-28 00:00:34,930 AppInfoParser [INFO] Kafka version : 0.10.1.0
2018-06-28 00:00:34,930 AppInfoParser [INFO] Kafka commitId : 3402a74efb23d1d4
2018-06-28 00:00:34,931 WARN NetworkClient [Thread-40-KafkaSpout-executor[12 12]] Bootstrap broker ip1:9092 disconnected
2018-06-28 00:00:35,092 WARN NetworkClient [Thread-40-KafkaSpout-executor[12 12]] Bootstrap broker ip2:9092 disconnected
2018-06-28 00:00:35,251 WARN NetworkClient [Thread-40-KafkaSpout-executor[12 12]] Bootstrap broker ip3:9092 disconnected
2018-06-28 00:00:35,524 WARN NetworkClient [Thread-40-KafkaSpout-executor[12 12]] Bootstrap broker ip4:9092 disconnected
2018-06-28 00:00:35,629 WARN NetworkClient [Thread-40-KafkaSpout-executor[12 12]] Bootstrap broker ip5:9092 disconnected
2018-06-28 00:00:35,822 WARN NetworkClient [Thread-40-KafkaSpout-executor[12 12]] Bootstrap broker ip6:9092 disconnected
2018-06-28 00:00:35,927 WARN NetworkClient [Thread-40-KafkaSpout-executor[12 12]] Bootstrap broker ip7:9092 disconnected


Here is the code of Kafka Spout:

private def getKafkaSpoutConfig(source: TopologyConfig) = {
System.clearProperty("java.security.auth.login.config") //tried this after getting error, no impact
KafkaSpoutConfig.builder("ip1:9092,ip2:9092,ip3:9092,.....,ip10:9092", "topicName")
.setProp(ConsumerConfig.GROUP_ID_CONFIG, "myConsumerGroup")
.setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer")
.setOffsetCommitPeriodMs(100)
.setFirstPollOffsetStrategy(KafkaSpoutConfig.FirstPollOffsetStrategy.UNCOMMITTED_LATEST)
.setMaxUncommittedOffsets(1000000)
.build()
}
def getKafkaSpout(source: TopologyConfig) = new KafkaSpout(getKafkaSpoutConfig(source: TopologyConfig))


On debugging, I see following is the stack trace for this error:

maybeHandleDisconnection:568, NetworkClient$DefaultMetadataUpdater (org.apache.kafka.clients)
processDisconnection:396, NetworkClient (org.apache.kafka.clients)
handleDisconnections:464, NetworkClient (org.apache.kafka.clients)
poll:270, NetworkClient (org.apache.kafka.clients)
poll:232, ConsumerNetworkClient (org.apache.kafka.clients.consumer.internals)
poll:195, ConsumerNetworkClient (org.apache.kafka.clients.consumer.internals)
getTopicMetadata:253, Fetcher (org.apache.kafka.clients.consumer.internals)
partitionsFor:1318, KafkaConsumer (org.apache.kafka.clients.consumer)
getFilteredTopicPartitions:57, NamedTopicFilter (org.apache.storm.kafka.spout)
refreshAssignment:54, ManualPartitionSubscription (org.apache.storm.kafka.spout)
subscribe:49, ManualPartitionSubscription (org.apache.storm.kafka.spout)
subscribeKafkaConsumer:657, KafkaSpout (org.apache.storm.kafka.spout)
activate:648, KafkaSpout (org.apache.storm.kafka.spout)
invoke:484, util$async_loop$fn__557 (org.apache.storm)
run:22, AFn (clojure.lang)
run:748, Thread (java.lang)


The same code was working for one Kafka setup, but for another setup of kafka of same version it started to give above error.

As par comments, I tried to connect to port 9092 of Kafka, Which I was able to do:

➜ git:(myBranch) ✗ telnet ipn 9092
Trying ipn...
Connected to my-kafka-app-396433.
Escape character is '^]'.


conf.setDebug(true)

1 Answer
1

This was happening because of some version mismatch of kafka. The installed kafka version was 0.10.0.1 while the code was picking and executing with kafka-clients version: 0.10.1.0.

0.10.0.1

0.10.1.0

It was happening as storm-core has a dependency of kafka-clients version: 0.10.1.0, which can be overwritten, which I did but somehow it was not excluded properly in sbt. After some permutations it was working and final dependencies were looking like this:

storm-core

kafka-clients

0.10.1.0

Seq("org.apache.storm" % "storm-core" % "1.2.1" % "compile" excludeAll(
ExclusionRule(organization = "org.apache.logging.log4j"),
ExclusionRule(organization = "org.apache.kafka", artifact = "kafka-clients"),
ExclusionRule(organization = "ring-cors"),
ExclusionRule(organization = "org.slf4j", artifact = "*"),
ExclusionRule(organization = "log4j", artifact = "*"),
ExclusionRule(organization = "javax", artifact = "*"),
ExclusionRule(organization = "javax.servlet", artifact="*")
)),
"org.apache.kafka" % "kafka-clients" % "0.10.0.1" excludeAll(
ExclusionRule(organization = "org.slf4j", name = "slf4j-log4j12"),
ExclusionRule(organization = "log4j"),
ExclusionRule(organization = "javax.servlet", artifact="*")
),
"org.apache.storm" % "storm-kafka-client" % "1.2.1" excludeAll(
ExclusionRule(organization = "org.apache.kafka", artifact = "*"),
ExclusionRule(organization = "org.apache.logging.log4j"),
ExclusionRule(organization = "ring-cors"),
ExclusionRule(organization = "org.apache.logging.log4j", artifact = "*"),
ExclusionRule(organization = "org.slf4j", artifact = "*"),
ExclusionRule(organization = "log4j", artifact = "*")
),


By clicking "Post Your Answer", you acknowledge that you have read our updated terms of service, privacy policy and cookie policy, and that your continued use of the website is subject to these policies.