Before MergeContent you can put an UpdateAttribute processor and create a new attribute "xx_yy" with the value of ${xx}_${yy} then in MergeContent put xx_yy as the correlation attribute. ... View more

The documentation for the processor is here: https://nifi.apache.org/docs/nifi-docs/components/org.apache.nifi/nifi-kafka-0-9-nar/1.6.0/org.apache.nifi.processors.kafka.pubsub.PublishKafka/index.html It says... Dynamic Properties: Dynamic Properties allow the user to specify both the name and value of a property. The name of a Kafka configuration property. The value of a given Kafka configuration property. These properties will be added on the Kafka configuration after loading any provided configuration properties. In the event a dynamic property represents a property that was already set, its value will be ignored and WARN message logged. For the list of available Kafka properties please refer to: http://kafka.apache.org/documentation.html#configuration. ... View more

There should be a property called "Message Key Field" which is the name of the field in the records that you want to use as the message key. It depends how your CsvReader is configured... if it is creating a schema from the column headers then this value would be the name of the column, or if you are providing a specific schema then it would be the name of the field from the schema your provided. ... View more

ListSFTP writes two attributes on each flow file - filename and path. In FetchSFTP you want to set the remote file to be a dynamic expression that uses these attributes: ${path}/${filename} ... View more

The queue before the WW processor shows 0 flow files queued so there is nothing to pull. Also, since the queue between XX and ZZ is showing that it is full, this means back-pressure is engaged and the XX processor will no longer execute until space is available in that queue, which then has the side-effect that XX also can't put more flow files into the queue before WW. ... View more