question Re: NiFi GetKafka ZooKeeper Connection Error in Archives of Support Questions (Read Only)

NiFi GetKafka ZooKeeper Connection Error

stephmm30 — Wed, 27 Jul 2016 05:58:30 GMT

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

Re: NiFi GetKafka ZooKeeper Connection Error

bbende — Wed, 27 Jul 2016 09:18:04 GMT

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.

Re: NiFi GetKafka ZooKeeper Connection Error

stephmm30 — Thu, 28 Jul 2016 00:45:50 GMT

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

Re: NiFi GetKafka ZooKeeper Connection Error

bbende — Thu, 28 Jul 2016 01:00:25 GMT

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-Kafka-client-code-td9098.html

Re: NiFi GetKafka ZooKeeper Connection Error

stephmm30 — Thu, 28 Jul 2016 02:01:20 GMT

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

Re: NiFi GetKafka ZooKeeper Connection Error

bbende — Thu, 28 Jul 2016 02:29:48 GMT

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.

Re: NiFi GetKafka ZooKeeper Connection Error

stephmm30 — Thu, 28 Jul 2016 02:50:15 GMT

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

Re: NiFi GetKafka ZooKeeper Connection Error

bbende — Thu, 28 Jul 2016 03:07:12 GMT

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.

Re: NiFi GetKafka ZooKeeper Connection Error

stephmm30 — Thu, 28 Jul 2016 05:40:04 GMT

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?

Re: NiFi GetKafka ZooKeeper Connection Error

bbende — Thu, 28 Jul 2016 07:53:08 GMT

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.

Re: NiFi GetKafka ZooKeeper Connection Error

stephmm30 — Fri, 05 Aug 2016 04:00:33 GMT

Hi Bryan,

I've successfully get data from Kafka and save it down to HDFS in a one master one slave cluster. When I added one more slave node restart my data flow, however, only the first slave works, the newly added slave is live. When I increase the concurrent task in ConsumeKafka from 1 to 5, which is actually how many partition we have in the Kafka cluster, I get error saying 'kafkaconsumer is not safe for multi-threaded access'. Do I need set up a ConsumeKafka processor group in this circumstance to allow this data flow running in parallel across the whole NiFi cluster?

Thanks for the help,

Stephanie

Re: NiFi GetKafka ZooKeeper Connection Error

bbende — Fri, 05 Aug 2016 06:14:11 GMT

Hi Stephanie, I'm actually not sure about that one, but I think it has more to do with the Kafka client and the number of partitions. You should be able to have a ConsumeKafka processor running on each node of your NiFi cluster and each pulling data without doing anything special. It might be good to start a new question about this specific problem with ConsumeKafka only consuming data on one node.