NiFi GetKafka ZooKeeper Connection Error

[#1]

Hi,

I'm trying to set up data flow by using a GetKafka processor to pull data from a Kafka cluster and a PutFile to save it on local file. However, I ran into connection refused error even though my server has access to the ZooKeeper connection by using telnet. (The log shows connected first and then refused...). I'm on NiFi 0.7 and Kafka 0.9, so it shouldn't be a version issue. Can anyone please help out?

org.I0Itec.zkclient.ZkClient zookeeper state changed (SyncConnected)
2016-07-26 21:15:09,690 INFO [pool-67-thread-1-SendThread(52.90.171.224:2181)] org.apache.zookeeper.ClientCnxn Opening socket connection to server 52.90.171.224/52.90.171.224:2181. Will not attempt to authenticate using SASL (unknown error)
2016-07-26 21:15:09,693 INFO [pool-67-thread-1-SendThread(52.90.171.***:2181)] org.apache.zookeeper.ClientCnxn Socket connection established to 52.90.171.224/52.90.171.224:2181, initiating session
2016-07-26 21:15:09,698 INFO [pool-67-thread-1-SendThread(52.90.171.224:2181)] org.apache.zookeeper.ClientCnxn Session establishment complete on server 52.90.171.224/52.90.171.224:2181, sessionid = 0x2e560467decc2fed, negotiated timeout = 6000
2016-07-26 21:15:09,699 INFO [pool-67-thread-1-EventThread] org.I0Itec.zkclient.ZkClient zookeeper state changed (SyncConnected)
2016-07-26 21:15:09,707 INFO [Framework Task Thread Thread-4-SendThread(localhost:2181)] org.apache.zookeeper.ClientCnxn Opening socket connection to server localhost/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error)
2016-07-26 21:15:09,708 WARN [Framework Task Thread Thread-4-SendThread(localhost:2181)] org.apache.zookeeper.ClientCnxn Session 0x0 for server null, unexpected error, closing socket connection and attempting reconnect
java.net.ConnectException: Connection refused
        at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) ~[na:1.7.0_80]
        at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:744) ~[na:1.7.0_80]
        at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:361) ~[na:na]
        at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1081) ~[na:na]

Thanks!

Stephanie

[#2]

It looks like there are two different connections being attempted here... the first one is to 52.90.171.224:2181 and then the one where the error is coming from is to localhost:2181.

Can you confirm if 52.90.171.224 is a remote server, or is it the same IP of the machine where NiFi is running?

Also, are you running a NiFi cluster or single instance? and if running a cluster, are you running an embedded ZooKeeper for NiFi's state management? Just trying to see if ZooKeeper is being used for anything else here.

[#3]

Hi Bryan,

Thanks for the reply! I'm running NiFi cluster with no embedded ZooKeeper. We have stand alone ZooKeeper installed on the cluster already, and I put this ZooKeeper name in Nifi's zookeeper.properties file. 52.90.171.224 is a remote Kafka node, not my NiFi slave node.

After adding ZooKeeper configuration mentioned above, I restarted and get some other error saying 'o.apache.nifi.processors.kafka.GetKafka java.lang.IllegalStateException: java.util.concurrent.ExecutionException: java.lang.AssertionError: assertion failed'. Full stack as below. Any thought?

2016-07-27 17:14:57,110 ERROR [Timer-Driven Process Thread-5] o.apache.nifi.processors.kafka.GetKafka java.lang.IllegalStateException: java.util.concurrent.ExecutionException: java.lang.AssertionError: assertion failed at org.apache.nifi.processors.kafka.GetKafka.onTrigger(GetKafka.java:355) ~[na:na] at org.apache.nifi.processor.AbstractProcessor.onTrigger(AbstractProcessor.java:27) ~[nifi-api-0.7.0.jar:0.7.0] at org.apache.nifi.controller.StandardProcessorNode.onTrigger(StandardProcessorNode.java:1054) [nifi-framework-core-0.7.0.jar:0.7.0] at org.apache.nifi.controller.tasks.ContinuallyRunProcessorTask.call(ContinuallyRunProcessorTask.java:136) [nifi-framework-core-0.7.0.jar:0.7.0] at org.apache.nifi.controller.tasks.ContinuallyRunProcessorTask.call(ContinuallyRunProcessorTask.java:47) [nifi-framework-core-0.7.0.jar:0.7.0] at org.apache.nifi.controller.scheduling.TimerDrivenSchedulingAgent$1.run(TimerDrivenSchedulingAgent.java:127) [nifi-framework-core-0.7.0.jar:0.7.0] at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471) [na:1.7.0_80] at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:304) [na:1.7.0_80] at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:178) [na:1.7.0_80] at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) [na:1.7.0_80] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) [na:1.7.0_80] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) [na:1.7.0_80] at java.lang.Thread.run(Thread.java:745) [na:1.7.0_80] Caused by: java.util.concurrent.ExecutionException: java.lang.AssertionError: assertion failed at java.util.concurrent.FutureTask.report(FutureTask.java:122) [na:1.7.0_80] at java.util.concurrent.FutureTask.get(FutureTask.java:202) [na:1.7.0_80] at org.apache.nifi.processors.kafka.GetKafka.onTrigger(GetKafka.java:348) ~[na:na] ... 12 common frames omitted Caused by: java.lang.AssertionError: assertion failed at scala.Predef$.assert(Predef.scala:165) ~[na:na] at kafka.consumer.TopicCount$anonfun$makeConsumerThreadIdsPerTopic$2.apply(TopicCount.scala:51) ~[na:na] at kafka.consumer.TopicCount$anonfun$makeConsumerThreadIdsPerTopic$2.apply(TopicCount.scala:49) ~[na:na] at scala.collection.TraversableLike$WithFilter$anonfun$foreach$1.apply(TraversableLike.scala:772) ~[na:na] at scala.collection.immutable.Map$Map1.foreach(Map.scala:109) ~[na:na] at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771) ~[na:na] at kafka.consumer.TopicCount$.makeConsumerThreadIdsPerTopic(TopicCount.scala:49) ~[na:na] at kafka.consumer.StaticTopicCount.getConsumerThreadIdsPerTopic(TopicCount.scala:113) ~[na:na] at kafka.consumer.ZookeeperConsumerConnector.consume(ZookeeperConsumerConnector.scala:226) ~[na:na] at kafka.javaapi.consumer.ZookeeperConsumerConnector.createMessageStreams(ZookeeperConsumerConnector.scala:85) ~[na:na] at kafka.javaapi.consumer.ZookeeperConsumerConnector.createMessageStreams(ZookeeperConsumerConnector.scala:97) ~[na:na] at org.apache.nifi.processors.kafka.GetKafka.createConsumers(GetKafka.java:281) ~[na:na] at org.apache.nifi.processors.kafka.GetKafka$1.call(GetKafka.java:343) ~[na:na] at org.apache.nifi.processors.kafka.GetKafka$1.call(GetKafka.java:340) ~[na:na] at java.util.concurrent.FutureTask.run(FutureTask.java:262) [na:1.7.0_80] ... 3 common frames omitted

Thanks,

Stephanie

[#4]

Glad we got past the first problem... out of curiosity, if you created another new GetKafka processor pointing at the same Kafka topic, does it also get the same error?

I was reading this old thread that seemed related, and it hinted that some kind of bad state might be stuck in ZooKeeper:

http://apache-nifi-developer-list.39713.n7.nabble.com/GetKafka-blowing-up-with-assertion-error-in-Ka...

[#5]

Hi Bryan,

I looked closely at the log and found my processor actually established connection to Kafka client. But it retrieved 0 partition for my topic, which is supposed to be 6 instead. Then it push 0 as concurrentTaskToUse to the topicCountMap, which I guess resulted in the assertion failed error above.

Is there a way I can specify how many partition I should retrieve or any way I can debug to let KafkaUtils get the correct Kafka partition number?

Thanks,

Stephanie

[#6]

Hi Stephanie,

Were you able to try creating another GetKafka processor for the same topic on your NiFi graph to see if it experiences the same problem?

The reason I wanted to try this was because each GetKafka processor has a property for Client Name which is specific to the processor's id, it ends up being something like "NiFi-<uuid>". This id will end up being stored in ZooKeeper somewhere to identify the consumer.

So I wanted to see if creating a new GetKafka processor, and thus a new client id, would get the same problem or not.

[#7]

Hi Bryan,

Yes, I created a new processor with the same setting, but still get 0 partition warning and the same error. Any thoughts?

Thanks,

Stephanie

[#8 · Accepted solution]

I just realized something... GetKafka and PutKafka use the kafka-client 0.8.2, and there are newer processors in 0.7.0 called ConsumeKafka and PublishKafka which use kafka-client 0.9.0.1.

Since you are using Kafka 0.9 I think we should be using ConsumeKafka here. Lets see if that works any better, and sorry for the confusion.

[#9]

Hi Bryan,

Thanks for pointing this out! I switched to ConsumerKafka and now everything gets work. One more question thought, does NiFi support CName or ARecord as broker instead of a list of real ips, since this is really import for us?

[#10]

Glad to hear it is working. As far as the broker, NiFi is not doing anything special, it just takes the value entered in the processor config and passes it to Kafka as the property ProducerConfig.BOOTSTRAP_SERVERS_CONFIG (which is "bootstrap.servers"). So it really comes down to what Kafka does with this list. I would think it can be any hostname or IP that can be resolved.