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 ... View more

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? ... View more

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 ... View more

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 ... View more

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 ... View more