Hi All, I need a processor which reads a csv file and save the data into database (Amazon Aurora rds) Initially i had built it using standard processors : csv-to-avro , avro-to-json ,splitjson, json-to-sql and finally Putsql This is running fine, however we also need some additional stats now to be saved in other table like, when a file processing was it started, when it end and how many rows it saved from total rows in source file. Which I am not able to link with standard processor as PutSQL is working row by row, ie. taking insert sql as one flow file. So as the file would keep coming from source stream. We would not know when 1 file is finished processing and mainatain the count of success rows. Planning to built a custom processor for this , Any suggestion on how shall we approch this in nifi or rather trigger some spark/custom process to do this by passing file from nifi. Thanks Tarun ... View more

I am facing a memory issue with a specific processor ConvertJSONToSQL Not sure if there is a way to configure processor specific memory, as when this error comes i found other processor and services running fine in nifi. Is there a way to address this apart from looking extending heapspace in general to whole nifi , currently its 8GB. Below is the stack trace of error: 2018-02-09 05:42:53,197 ERROR [Timer-Driven Process Thread-1] o.a.n.p.standard.ConvertJSONToSQL ConvertJSONToSQL[id=6f1f8d3a-20b3-3fc5-884e-73f1e7443b58] ConvertJSONToSQL[id=6f1f8d3a-20b3-3fc5-884e-73f1e7443b58] failed to process session due to java.lang.OutOfMemoryError: Java heap space: {} java.lang.OutOfMemoryError: Java heap space at java.util.Arrays.copyOf(Arrays.java:3236) at java.io.ByteArrayOutputStream.grow(ByteArrayOutputStream.java:118) at java.io.ByteArrayOutputStream.ensureCapacity(ByteArrayOutputStream.java:93) at java.io.ByteArrayOutputStream.write(ByteArrayOutputStream.java:153) at java.io.DataOutputStream.writeLong(DataOutputStream.java:224) at org.apache.nifi.repository.schema.SchemaRecordWriter.writeFieldValue(SchemaRecordWriter.java:117) at org.apache.nifi.repository.schema.SchemaRecordWriter.writeFieldRepetitionAndValue(SchemaRecordWriter.java:69) at org.apache.nifi.repository.schema.SchemaRecordWriter.writeRecordFields(SchemaRecordWriter.java:56) at org.apache.nifi.repository.schema.SchemaRecordWriter.writeRecordFields(SchemaRecordWriter.java:47) at org.apache.nifi.repository.schema.SchemaRecordWriter.writeFieldValue(SchemaRecordWriter.java:147) at org.apache.nifi.repository.schema.SchemaRecordWriter.writeFieldRepetitionAndValue(SchemaRecordWriter.java:96) at org.apache.nifi.repository.schema.SchemaRecordWriter.writeRecordFields(SchemaRecordWriter.java:56) at org.apache.nifi.repository.schema.SchemaRecordWriter.writeRecordFields(SchemaRecordWriter.java:47) at org.apache.nifi.repository.schema.SchemaRecordWriter.writeFieldValue(SchemaRecordWriter.java:143) at org.apache.nifi.repository.schema.SchemaRecordWriter.writeFieldRepetitionAndValue(SchemaRecordWriter.java:69) Thanks, Tarun Kumar ... View more

We have been facing an intermittent issue in our QA env, where PutHDFS processor goes stuck and not able to release item from upstream processor queue. This does not look related to load as the no.of messages in queue and queue size was not too high. We tried to stop/start the processor but this does not work as after stopping the processor we don't see the start button on PutHDFS. The only way we are currently fixing this is by restarting the nifi which we would like to avoid in Prod env. At the time of stuck No of threads we can see on UI 2 and processor is configured with default 1 parallel level. In order to further drill down we tried to take multiple thread-dumps and notices that a particular thread was always blocked with same stack trace and linked to PutHDFS processor. Stack trace is as below. Timer-Driven Process Thread-2" Id=106 BLOCKED on java.io.BufferedInputStream@b7146b3 at java.io.BufferedInputStream.read(BufferedInputStream.java:336) at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:284) at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:326) ------- at org.apache.nifi.processors.hadoop.PutHDFS$1.run(PutHDFS.java:255) at java.security.AccessController.doPrivileged(Native Method) puthdfs-stuck.jpgthread-dump.txtthread-dump-2.txtthread-dump-3.txt I am attaching the UI screen-shot at time of stuck and thread-dumps for reference. Can someone please help us in finding the root-cause for this ? ... View more

Hi Team, we noticed ListS3 processor is processing same file twice or more, while there was no change in the source file. As we go though the "Component State" which clearly says, nifi manages the file states with latest time-stamp and list only when a new file has been added or existing file has been modified. We are seeing ListS3 processor is reprocessing same file from Data Provenance which has the same s3.lastModifed property value, We believe file should not be re-processed unless its changed per "State Management" docs. Kindly help us on this issue. Let us know for any information needed. Note: ListS3 Processor has been setup to run only primary node. ... View more

Hi Team, I have built a flow in nifi, where reading files from hdfs and we need to trigger spark job by passing these files. We are using Livy to trigger spark job and facing some issue while PostHTTPProcessor is triggering livy URL. Below are some details. 1) PostHTTP, We give a URL to our Livy server 2)Passing parameters for filename as args 3)Passing spark-job jar as file: /jar/singledataplatform/sd-spark-1.0-SNAPSHOT.jar 4)Passing className: the fully qualified classname for sparkjob i.e. com.markit.sdp.spark.SparkApplication We get 500 server error in the nifi-app.log as below 2017-10-12 10:45:57,733 ERROR [Timer-Driven Process Thread-7] o.a.nifi.processors.standard.PostHTTP PostHTTP[id=eba83476-015e-1000-7b4f-a5b82838282f] Failed to Post StandardFlowFileRecord[uuid=d86a6ff2-c8d5-4d53-af1d-2e7f6f7da000,claim=StandardContentClaim [resourceClaim=StandardResourceClaim[id=1507819557354-1156, container=default, section=132], offset=757275, length=197],offset=0,name=PV_170_Options_L1600_2017-10-04T131500-0400.txt.gz,size=197] to https://aw1frtdvhdp01.mdevlab.com:8443/gateway/default/livy/v1/batches: response code was 500:Server Error On checking the livy logs we see below line. "POST /batches?doAs=pv.hdp-user HTTP/1.1" 400 - The knox, gateway.log are having exception as below. 2017-10-12 15:45:57,727 INFO hadoop.gateway (AclsAuthorizationFilter.java:doFilter(85)) - Access Granted: true 2017-10-12 15:45:57,732 ERROR hadoop.gateway (AbstractGatewayFilter.java:doFilter(69)) - Failed to execute filter: java.util.EmptyStackException 2017-10-12 15:45:57,733 ERROR hadoop.gateway (AbstractGatewayFilter.java:doFilter(66)) - Failed to execute filter: javax.servlet.ServletException: java.util.EmptyStackException 2017-10-12 15:45:57,733 ERROR hadoop.gateway (AbstractGatewayFilter.java:doFilter(66)) - Failed to execute filter: javax.servlet.ServletException: org.apache.shiro.subject.ExecutionException: java.security.PrivilegedActionException: javax.servlet.ServletException: java.util.EmptyStackException 2017-10-12 15:45:57,733 ERROR hadoop.gateway (AbstractGatewayFilter.java:doFilter(66)) - Failed to execute filter: javax.servlet.ServletException: org.apache.shiro.subject.ExecutionException: java.security.PrivilegedActionException: javax.servlet.ServletException: java.util.EmptyStackException 2017-10-12 15:45:57,733 ERROR hadoop.gateway (GatewayFilter.java:doFilter(145)) - Gateway processing failed: javax.servlet.ServletException: org.apache.shiro.subject.ExecutionException: java.security.PrivilegedActionException: javax.servlet.ServletException: java.util.EmptyStackException javax.servlet.ServletException: org.apache.shiro.subject.ExecutionException: java.security.PrivilegedActionException: javax.servlet.ServletException: java.util.EmptyStackException at org.apache.shiro.web.servlet.AdviceFilter.cleanup(AdviceFilter.java:196) at org.apache.shiro.web.filter.authc.AuthenticatingFilter.cleanup(AuthenticatingFilter.java:155) at org.apache.shiro.web.servlet.AdviceFilter.doFilterInternal(AdviceFilter.java:148) at org.apache.shiro.web.servlet.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:125) at org.apache.shiro.web.servlet.ProxiedFilterChain.doFilter(ProxiedFilterChain.java:66) at org.apache.shiro.web.servlet.AbstractShiroFilter.executeChain(AbstractShiroFilter.java:449) at org.apache.shiro.web.servlet.AbstractShiroFilter$1.call(AbstractShiroFilter.java:365) at org.apache.shiro.subject.support.SubjectCallable.doCall(SubjectCallable.java:90) ... View more

We are trying to setup Kafka with kerberos and facing some issue in running defaultproducer/consumer example. Taken help from : https://community.hortonworks.com/articles/55154/kafka-ssl-kerberos-cheat-sheet-settingsconsole-com.html [tarun.kumar@aw1frtdvhdp02 bin]$ /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh --broker-list localhost:6667 --topic test Hello [2017-05-29 14:11:56,293] ERROR Error when sending message to topic test with key: null, value: 5 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) While trying to run the producer/consumer above ways seeing above error. Can anyone please help me in understanding /resolving this ? ... View more