NiFi docs mention this: While the contents and attributes of a FlowFile can change, the FlowFile object is immutable. How is flowfile immutable if its contents and attributes can change? ... View more

Is this the correct way of exposing a nifi property as env variable(in start.sh):   prop_replace 'nifi.web.http.port' "${NIFI_WEB_HTTP_PORT:-8080}"  I expect it to expose nifi.web.http.port as NIFI_WEB_HTTP_PORT, and also setting it to default value of 8080. ... View more

Here is my unit test code: runner.setProperty(....); RecordSchema inputRecordSchema = ....create schema based on RecordField final Object[] recordArray = recordArray(inputRecordSchema); // create records(instances of MapRecord) parser.addRecord((Object) recordArray); .... Object[] recordArray(RecordSchema schema){ Record record = new MapRecord(schema, new HashMap<String, Object> { { put(...) ... ... } } ..... } In main code, I iterate records from reader: while((record = reader.nextRecord()) != null) { //process record } But I see that record, though MapRecord instance, is empty. Seems I am missing some step in unit test. How to sort it out? ... View more

I am trying to implement for my custom processor: @Override public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException Basically I do: FlowFile original = session.get(); ... ... try (final InputStream in = session.read(original); final RecordReader reader = readerFactory.createRecordReader(original, in, getLogger())) { Record inputRecord = null; output = session.create(original); while((inputRecord = reader.nextRecord()) != null) { final WriteResult writeResult; ... ... In unit test, I have this arrangement: List<RecordField> fields = new ArrayList<>(); RecordField idField = new RecordField("id", RecordFieldType.STRING.getDataType()); fields.add(idField); .... more fields RecordSchema inputRecordSchema = new SimpleRecordSchema(fields); RecordField inputRecord = new RecordField("inputRecord", RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.RECORD.getRecordDataType(inputRecordSchema))); parser.addSchemaField(inputRecord); final Record record1 = new MapRecord(inputRecordSchema, new HashMap<String, Object>(){....}); final Record record2 = new MapRecord(inputRecordSchema, new HashMap<String, Object>() {....}); final Object[] recordArray = new Object[] { record1, record2 }; parser.addRecord((Object) recordArray); Though I explicitly set inputRecord, I expect that reader.nextRecord() in main code would fetch me inner records one by one. So first I would get record1, then record2. But instead I get parent record(by name inputRecord). As a result, I need to write explicit code to parse the record, something like: Arrays.asList(inputRecord.getValues()).get(0).... I want to get rid of this. Any input in improving this would help. ... View more