@Shu Thank you for the response. It is working fine without any errors. However, when I go to the dynamodb table, the entire JSON object from the FlowFile lands under the property set in JSON Document Attribute (in this case, activity) for the PutDynamoDB processor. Attached is a screenshot of the table in dynamodb (3.jpg) 3.jpg Sample JSON Input: { "messageid": "1", "application": "value 1", "datetime": "value 2", "dev": "value 3", "activity": "value 4" } Sample Output in DynamoDB table under a single field called 'activity': { "application" : { "S" : "value 1" }, "datetime" : { "S" : "value 2" }, "dev" : { "S" : "value 3" }, "activity" : { "S" : "value 4" }, "messageid" : { "S" : "1" } } I want the JSON contents of the FlowFile to land in DynamoDB as a record with a messageid and 4 fields with their respective values. I don't know if this is a bug or if I am using it incorrectly. Is there any workaround? Thanks in advance! ... View more

@Shu Thank you for the response. It is working fine without any errors. However, when I go to the dynamodb table, the entire JSON object from the FlowFile lands under the property set in JSON Document Attribute (in this case, activity) for the PutDynamoDB processor. Attached is a screenshot of the table in dynamodb (3.jpg) 3.jpg Sample JSON Input: { "messageid": "1", "application": "value 1", "datetime": "value 2", "dev": "value 3", "activity": "value 4" } Sample Output in DynamoDB table under a single field called 'activity': { "application" : { "S" : "value 1" }, "datetime" : { "S" : "value 2" }, "dev" : { "S" : "value 3" }, "activity" : { "S" : "value 4" }, "messageid" : { "S" : "1" } } I want the JSON contents of the FlowFile to land in DynamoDB as a record with a messageid and 4 fields with their respective values. I don't know if this is a bug or if I am using it incorrectly. Is there any workaround? Thanks in advance! ... View more

@Shu Thank you for the response. It is working fine without any errors. However, when I go to the dynamodb table, the entire JSON object from the FlowFile lands under the property set in JSON Document Attribute (in this case, activity). JSON Input: { "messageid": "1", "application": "value 1", "datetime": "value 2", "dev": "value 3", "activity": "value 4" } For example, I have the below in a field called activity in dynamodb table. Attached is a screenshot of the table in dynamodb (3.jpg) Output in DynamoDB table under a field called 'activity': { "application" : { "S" : "value 1" }, "datetime" : { "S" : "value 2" }, "dev" : { "S" : "value 3" }, "activity" : { "S" : "value 4" }, "messageid" : { "S" : "1" } } I want the JSON contents of the FlowFile to land in DynamoDB as a record with a messageid and 4 fields with their respective values. I don't know if this is a bug or if I am using it incorrectly. Is there any workaround? Thanks! 3.jpg ... View more

@Wynner Thanks for the suggestion, I have not tried this yet but will have to check with the server administrator to see if this is something they don't mind trying. Jason ... View more

@Vinicius Higa Murakami - Thanks for the suggestion. I looked at the post you sent and tried to access some of the steps but it seems that I don't have the zk.cli.sh file that was suggested. I am running zookeeper as part of a Kafka download and it does not have this file. We have a current project that is utilizing Kafka and I decided to try connecting from there as well but no luck either. Jason ... View more

@Vinicius Higa Murakami -- I just tried the command again with the zookeeper parameter and got the following: GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG OWNER Could not fetch offset from kafka for group console-consumer-96628 partition [test,0] due to org.apache.kafka.common.errors.NotCoordinatorForGroupException: This is not the correct coordinator for this group.. console-consumer-96628 test 0 unknown 56 unknown none Any thoughts you have on this output would be greatly appreciated. Thanks again for your help! Jason ... View more

@Vinicius Higa Murakami -- So I tried the first command and it returned console-consumer-96628. I then tried to use that name in the second command like this but think I got something wrong because it doesn't accept it: bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group console-consumer-96628. I was also talking to a co-worker who uses Kafka more and we were wondering if I have to create a consumer group and then assign my topic to that group. Then on the Nifi end I would have to supply that proper Group Id in the processor? So far I usually just put some value like 0 in for the Group Id on the processor and hoped it would connect directly without actually needing the group. Thanks for your help! Jason ... View more

@Vinicius Higa Murakami -- Regarding your questions, I have posted my responses below: What about the port from zookeeper? (2181 by default). - Do you mean by using port 2181 instead of port 9092 in the Consume Kafka Processor? If so, I have open port 2181 and have changed the Kafka broker to ec2-XX-X-XXX-XXX.XX-XXXX-1.compute.amazonaws.com:2181 but still nothing came over. And could you check if there's any extra consumer group (probably used by Nifi Processor) - By extra consumer group do you mean another processor that is also trying to consume the same topic on that same port? If so, I can confirm that I have only one processor trying to access Kafka in this NiFi model. I had the same set up in NiFi 1.5 but that processor has been stopped and that instance of NiFi is not running. Kafka logs are showing something? - I have checked the Controller logs and the server logs and I only see the following: From the Controller Logs 2018-06-06 17:55:15,642] TRACE [Controller 0]: checking need to trigger partition rebalance (kafka.controller.KafkaController) [2018-06-06 17:55:15,642] DEBUG [Controller 0]: preferred replicas by broker Map(0 -> Map([test,0] -> List(0))) (kafka.controller.KafkaController) [2018-06-06 17:55:15,642] DEBUG [Controller 0]: topics not in preferred replica Map() (kafka.controller.KafkaController) [2018-06-06 17:55:15,643] TRACE [Controller 0]: leader imbalance ratio for broker 0 is 0.000000 (kafka.controller.KafkaController) I do not see any errors or warnings. Kafka is running on a Free Tier Linux within in Amazon Ec2 so I know there are restrictions of size and memory usages but since I was able to configure Kafka to use less memory it works fine within that local environment. I don't think that is the issue here but just thought to mention it. Jason ... View more

@Wynner - Nifi 1.6 is running on Windows Server 2016 hosted on Amazon Ec2 ... View more

@Matt Clarke - I am thinking I should ask this on another post but was curious what the proper kafka tool was for retrieving messages from a kafka topic, Consume or Get? Get looks like it's for older versions and I have been trying to use Consume_010 on Nifi 1.6. Maybe a compatibility issue between 1.6 and my kafka installation? Thanks, Jason ... View more

@Matt Clarke -- Thanks for the forum tip as well as the previous suggestions. I was able to get the ListenTCP working properly. Jason ... View more

Hey Matt, Thanks for the response. Yes, I am trying to receive data on a TCP port in Nifi. I have now switched the processor to Listen TCP and specified the port number that I want to listen on. I left the LAN option blank and I am now getting some new errors listed below: 2018-05-31 18:05:58,393 ERROR [Monitor Processore Lifecycle Thread-1] o.a.nifi.processors.standard.ListenTCP ListenTCP[id=b74d9e21-0163-1000-391b-4e77cd11cf16] Failed to properly initialize Processor. If still scheduled to run, NiFi will attempt to initialize and run the Processor again after the 'Administrative Yield Duration' has elapsed. Failure is due to java.lang.reflect.InvocationTargetException: java.lang.reflect.InvocationTargetException java.lang.reflect.InvocationTargetException: null Caused by: java.net.BindException: Address already in use: bind We have tested that the port is listening using netstat commands and it shows that only one listener is coming up so I am not sure what else could be binding the address? Is there a setting somewhere to get around this because if the data is pushing from a remote server to NiFi on this port wouldn't it essentially be connecting to that port? Thanks, Jason ... View more