NiFi notes

Unit-testing cribs

Stuff I don't have to keep in the pile that's my brain because I just look here (rather than deep down inside these chronologically recorded notes):

  @Test
  public void test() throws Exception
  {
    TestRunner runner = TestRunners.newTestRunner( new ExcerptBase64FromHl7() );

    final String HL7 =
         "MSH|^~\\&|Direct Email|XYZ|||201609061628||ORU^R01^ORU_R01|1A160906202824456947||2.5\n"
        + "PID|||000772340934||TEST^Sandoval^||19330801|M|||1 Mockingbird Lane^^ANYWHERE^TN^90210^||\n"
        + "PV1||O\n"
        + "OBR||62281e18-b851-4218-9ed5-bbf392d52f84||AD^Advance Directive"
        + "OBX|1|ED|AD^Advance Directive^3.665892.1.238973.3.1234||^multipart^^Base64^"
        + "F_ABCDEFGHIJKLMNOPQRSTUVWXYZ";

    // mock up some attributes...
    Map< String, String > attributes = new HashMap<>();
    attributes.put( "attribute-name", "attribute-value" );
    attributes.put( "another-name",   "another-value" );

    // mock up some properties...
    runner.setProperty( ExcerptBase64FromHl7.A_STATIC_PROPERTY, "static-property-value" );
    runner.setProperty( "dynamic property", "some value" );

    // add a dynamic property...
    runner.setValidateExpressionUsage( false );
    runner.setProperty( "property-name", "value" );

    // call the processor...
    runner.setValidateExpressionUsage( false );
    runner.enqueue( new ByteArrayInputStream( HL7.getBytes() ), attributes );
    runner.run( 1 );
    runner.assertQueueEmpty();

    // gather the results...
    List< MockFlowFile > results = runner.getFlowFilesForRelationship( ExcerptBase64FromHl7.SUCCESS );
    assertTrue( "1 match", results.size() == 1 );
    MockFlowFile result = results.get( 0 );

    String actual = new String( runner.getContentAsByteArray( result ) );

    assertTrue( "Missing binary content", actual.contains( "1_ABCDEFGHIJKLMNOPQRSTUVWXYZ" ) );

    // ensure the processor transmitted the attribute...
    String attribute = result.getAttribute( "attribute-name" );
    assertTrue( "Failed to transmit attribute", attribute.equals( "attribute-value" ) );
  }

Wednesday, 6 January 2016

Here is a collection of some NiFi notes...

Friday, 15 January 2016

Monday, 18 January 2016

Here's an example from NiFi Rocks:

package rocks.nifi.examples.processors;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;

import com.jayway.jsonpath.JsonPath;
import org.apache.commons.io.IOUtils;

import org.apache.nifi.annotation.behavior.SideEffectFree;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;

import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.logging.ProcessorLog;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;

/**
 * Our NiFi JSON processor from
 * http://www.nifi.rocks/developing-a-custom-apache-nifi-processor-json/
 */
@SideEffectFree
@Tags( { "JSON", "NIFI ROCKS" } )                       // for finding processor in the GUI
@CapabilityDescription( "Fetch value from JSON path." ) // also displayed in the processor selection box
public class JsonProcessor extends AbstractProcessor
{
  private List< PropertyDescriptor > properties;       // one of these for each property of the processor
  private Set< Relationship >        relationships;    // one of these for each possible arc away from the processor

  @Override public List< PropertyDescriptor > getSupportedPropertyDescriptors() { return properties; }
  @Override public Set< Relationship >        getRelationships()                { return relationships; }

  // our lone property...
  public static final PropertyDescriptor JSON_PATH = new PropertyDescriptor.Builder()
        .name( "Json Path" )
        .required( true )
        .addValidator( StandardValidators.NON_EMPTY_VALIDATOR )
        .build();

  // our lone relationship...
  public static final Relationship SUCCESS = new Relationship.Builder()
        .name( "SUCCESS" )
        .description( "Succes relationship" )
        .build();

  /**
   * Called at the start of Apache NiFi. NiFi is a highly multi-threaded environment; be
   * careful what is done in this space. This is why both the list of properties and the
   * set of relationships are set with unmodifiable collections.
   * @param context supplies the processor with a) a ProcessorLog, b) a unique
   *                identifier and c) a ControllerServiceLookup to interact with
   *                configured ControllerServices.
   */
  @Override
  public void init( final ProcessorInitializationContext context )
  {
    List< PropertyDescriptor > properties = new ArrayList<>();
    properties.add( JSON_PATH );
    this.properties = Collections.unmodifiableList( properties );

    Set< Relationship > relationships = new HashSet<>();
    relationships.add( SUCCESS );
    this.relationships = Collections.unmodifiableSet( relationships );
  }

  /**
   * Called when ever a flowfile is passed to the processor. The AtomicReference
   * is required because in order to have access to the variable in the call-back scope it
   * needs to be final. If it were just a final String it could not change.
   */
  @Override
  public void onTrigger( ProcessContext context, ProcessSession session ) throws ProcessException
  {
    final ProcessorLog              log   = this.getLogger();
    final AtomicReference< String > value = new AtomicReference<>();

    FlowFile flowfile = session.get();

    // ------------------------------------------------------------------------
    session.read( flowfile, new InputStreamCallback()
    {
      /**
       * Uses Apache Commons to read the input stream out to a string. Uses
       * JsonPath to attempt to read the JSON and set a value to the pass on.
       * It would normally be best practice in the case of a exception to pass
       * the original flowfile to a Error relation point.
       */
      @Override
      public void process( InputStream in ) throws IOException
      {
        try
        {
          String json   = IOUtils.toString( in );
          String result = JsonPath.read( json, "$.hello" );
          value.set( result );
        }
        catch( Exception e )
        {
          e.printStackTrace();
          log.error( "Failed to read json string." );
        }
      }
    } );
    // ------------------------------------------------------------------------

    String results = value.get();

    if( results != null && !results.isEmpty() )
      flowfile = session.putAttribute( flowfile, "match", results );

    // ------------------------------------------------------------------------
    flowfile = session.write( flowfile, new OutputStreamCallback()
    {
      /**
       * For both reading and writing to the same flowfile. With both the
       * outputstream call-back and stream call-back remember to assign it
       * back to a flowfile. This processor is not in use in the code, but
       * could have been. The example was deliberate to show a way of moving
       * data out of callbacks and back in.
       */
      @Override
      public void process( OutputStream out ) throws IOException
      {
        out.write( value.get().getBytes() );
      }
    } );
    // ------------------------------------------------------------------------

    /* Every flowfile that is generated needs to be deleted or transfered.
     */
    session.transfer( flowfile, SUCCESS );
  }

  /**
   * Called every time a property is modified in case we want to do something
   * like stop and start a Flowfile over in consequence.
   * @param descriptor indicates which property was changed.
   * @param oldValue value unvalidated.
   * @param newValue value unvalidated.
   */
  @Override
  public void onPropertyModified( PropertyDescriptor descriptor, String oldValue, String newValue ) { }
}

Tuesday, 19 January 2016

Lifecycle of a processor...

Before anything else, the processor's init() method is called. It looks like this happens at NiFi start-up rather than later when the processor is actually used.

The processor must expose to the NiFi framework all of the Relationships it supports. The relationships are the arcs in the illustration below.

Performing the work. This happens when onTrigger() is called (by the framework).

The onTrigger() method is passed the context and the session, and it must:

1. get the flowfile from the session,
2. perform a read of the flowfile stream,
3. obtain the results of reading,
4. put the attribute "match," on the session,
5. do work (whatever that is/however that is done),
6. perform a write of the modified flowfile stream,
7. note the success or failure of writing,
8. remove the flowfile, or
9. hand off the flowfile to the next processor.

A processor won't in fact be triggered, even if it has one or more FlowFile in its queue if the framework, checking downstream destinations (i.e. all other processors) and these are full. There is an annotation, @TriggerWhenAnyDestinationAvailable, which when present will allow the FlowFile to be processed as long as it can advance to the next processor.

@TriggerSerially will make the processor work as single-threaded. However, the processor implementation must still be thread-safe as the thread of execution may change.

Properties

Properties are exposed to the web interface via public method, but the properties themselves are locked using Collections.unmodifiableList() (if coded that way). If a property is changed as a processor is running, the latter can react eagerly if it implements onPropertyModified().

(There is no sample code of validating processor properties. Simply, the customValidate() method is called with a validation context and any that describe problems are returned . (But, is this implemented?)

  protected Collection< ValidationResult > customValidate( ValidationContext validationContext )
  {
    return Collections.emptySet();
  }

It's possible to augment initialization and enabling lifecycles through Java annotion. Some of these apply only to processor components, some to controller components, some to reporting tasks, to one or more.

• @OnAdded annotates methods to be called immediately after a processor's init() method. Such methods must take no argument.
• @OnEnabled annotates methods to be called immediate afer the controller service is enabled, i.e.: every time a user enables it. If the method throws an exception, a log message and bulletin will be issued. The service will lock into the ENABLING state and become unusable. Such methods may take no argument or a ConfigurationContext.
• @OnRemoved, invoked before a component is removed from the flow, must take no arguments.
• @OnScheduled/@OnUnscheduled, invoked every time a component is scheduled to run or not scheduled to run. No arguments.
• @OnShutdown, invoked whenever NiFI is shut down. No arguments.

Wednesday, 20 January 2016

Friday, 22 January 2016

Gathering resources (see dailies for today).

package com.etretatlogiciels.nifi.python.filter;

import org.apache.commons.io.IOUtils;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;

@Tags( { "example", "resources" } )
@CapabilityDescription( "This example processor loads a resource from the nar and writes it to the FlowFile content" )
public class WriteResourceToStream extends AbstractProcessor
{
  public static final Relationship REL_SUCCESS = new Relationship.Builder().name( "success" )
                                .description( "files that were successfully processed" )
                                .build();
  public static final Relationship REL_FAILURE = new Relationship.Builder().name( "failure" )
                                .description( "files that were not successfully processed" )
                                .build();

  private Set< Relationship > relationships;
  private String resourceData;

  @Override protected void init( final ProcessorInitializationContext context )
  {

    final Set< Relationship > relationships = new HashSet<>();
    relationships.add( REL_SUCCESS );
    relationships.add( REL_FAILURE );
    this.relationships = Collections.unmodifiableSet( relationships );

    final InputStream resourceStream = Thread.currentThread().getContextClassLoader().getResourceAsStream( "file.txt" );

    try
    {
      this.resourceData = IOUtils.toString( resourceStream );
    }
    catch( IOException e )
    {
      throw new RuntimeException( "Unable to load resources", e );
    }
    finally
    {
      IOUtils.closeQuietly( resourceStream );
    }
  }

  @Override
  public void onTrigger( final ProcessContext context, final ProcessSession session ) throws ProcessException
  {
    FlowFile flowFile = session.get();

    if( flowFile == null )
      return;

    try
    {
      flowFile = session.write( flowFile, new OutputStreamCallback()
      {

        @Override public void process( OutputStream out ) throws IOException
        {
          IOUtils.write( resourceData, out );
        }
      } );
      session.transfer( flowFile, REL_SUCCESS );
    }
    catch( ProcessException ex )
    {
      getLogger().error( "Unable to process", ex );
      session.transfer( flowFile, REL_FAILURE );
    }
  }
}

Monday, 25 January 2016

Tuesday, 26 January 2016

Today I was experiencing the difficulty of writing tests for a NiFi processor only to learn that there are helps from NiFi for this. I found the test for the original, JSON processor example.

package rocks.nifi.examples.processors;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.List;

import org.apache.commons.io.IOUtils;

import org.junit.Test;

import static org.junit.Assert.assertTrue;

import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.apache.nifi.util.MockFlowFile;

public class JsonProcessorTest
{
  @Test
  public void testOnTrigger() throws IOException
  {
    // Content to be mock a json file
    InputStream content = new ByteArrayInputStream( "{\"hello\":\"nifi rocks\"}".getBytes() );

    // Generate a test runner to mock a processor in a flow
    TestRunner runner = TestRunners.newTestRunner( new JsonProcessor() );

    // Add properites
    runner.setProperty( JsonProcessor.JSON_PATH, "$.hello" );

    // Add the content to the runner
    runner.enqueue( content );

    // Run the enqueued content, it also takes an int = number of contents queued
    runner.run( 1 );

    // All results were processed with out failure
    runner.assertQueueEmpty();

    // If you need to read or do aditional tests on results you can access the content
    List< MockFlowFile > results = runner.getFlowFilesForRelationship( JsonProcessor.SUCCESS );
    assertTrue( "1 match", results.size() == 1 );
    MockFlowFile result = results.get( 0 );
    String resultValue = new String( runner.getContentAsByteArray( result ) );
    System.out.println( "Match: " + IOUtils.toString( runner.getContentAsByteArray( result ) ) );

    // Test attributes and content
    result.assertAttributeEquals( JsonProcessor.MATCH_ATTR, "nifi rocks" );
    result.assertContentEquals( "nifi rocks" );
  }
}

What does it do? NiFi TestRunner mocks out the NiFi framework and creates all the underpinnings I was trying today to create. This makes it so that calls inside onTrigger() don't summarily crash. I stepped through the test (above) to watch it work. All the references to the TestRunner object are ugly and only to set up what the framework would create for the processor, but it does work.

Had to work through these problems...

java.lang.AssertionError: Processor has 2 validation failures:
'basename' validated against 'CCD_Sample' is invalid because 'basename' is not a supported property
'doctype' validated against 'XML' is invalid because 'doctype' is not a supported property

—these properties don't even exist.

Caused by: java.lang.IllegalStateException: Attempting to Evaluate Expressions but PropertyDescriptor[Path to Python Interpreter]
indicates that the Expression Language is not supported. If you realize that this is the case and do not want this error to occur,
it can be disabled by calling TestRunner.setValidateExpressionUsage(false)

—so I disabled it. However, when this error or something like it is issued in the debugger:

java.lang.AssertionError: Processor has 1 validation failures:
'property name' validated against 'property default value' is invalid because
'property name' is not a supported property

it's likely because you didn't add validation to the property when defined:

    .addValidator( StandardValidators.NON_EMPTY_VALIDATOR )

Anyway, let's keep going...

Then, I reached the session.read() bit which appeared to execute, but processing never returned. So, I killed it, set a breakpoint on the session.write(), and relaunched. It got there. I set a breakpoint on the while( ( line = reply.readLine() ) != null ) and it got there. I took a single step more and, just as before, processing never returned.

Then I break-pointed out.write() and the first call to session.putAttribute() and relaunched. Processing never hit those points.

Will continue this tomorrow.

Wednesday, 27 January 2016

Thursday, 28 January 2016

runner.assertQueueEmpty() will fail (and stop testing) if there are any flowfiles left untransferred in the queue. Did you forget to transfer a flowfile? Did you lose your handle to a flowfile because you used the variable to point at another flowfile before you had transferred it? Etc.

Thursday, 11 February 2016

NiFi logging example...

final ProcessorLog logger = getLogger();

String key   = "someKey";
String value = "someValue";
logger.warn( "This is a warning message" );
logger.info( "This is a message with some interpolation key={}, value={}", new String[]{ key, value } );

My second NiFi processor at work.

Tuesday, 16 February 2016

Documenting practical consumption of GetFile and PutFile processors (more or less "built-ins") of NiFi. These are the differences from default properties:

GetFile: Configure Processor

1. No auto-termination (Settings → Auto terminate relationships)
2. 30 seconds (Scheduling → Run schedule)
   This is how many seconds between GetFile pulsing and sending the (same) file again.
3. /tmp/nifi-source (Properties → Input Directory)
4. test-lily-munster.csv (Properties → File Filter)

PutFile: Configure Processor

1. Check failure and success (Settings → Auto terminate relationships)
   This is because I want to stop at PutFile in my example above—processing will not be going on to anywhere else.
2. /tmp/nifi-output (Properties → Directory)

My own AppointmentsProcessor: Configure Processor

1. Check FAILURE and ORIGINAL (Settings → Auto terminate relationships)
   This is because I want to stop on failure and do not wish to route original document on to the next point (PutFile).
2. East Bay Health (Properties → Provider)

The UI sequence appears to be:

1. Perform all configuration.
2. Any processor can be started because it will sit idle until something comes in or, if it's feeding into the pipeline and has something, it will just fill the pipeline which will await the next processor down-stream to start.

Monday, 22 February 2016

Looking to change my port number from 8080, the property is

nifi.web.http.port=8080

...and this is in ${NIFI_HOME}/conf/nifi.properties. Also of interest may be:

nifi.web.http.host
nifi.web.https.host
nifi.web.https.port

Monday, 29 February 2016

I'm testing actual NiFi set-ups for cda-breaker and cda-filter this morning. The former went just fine, but I'm going to document the setting up of a pipeline with the latter since I see I have not completely documented such a thing before.

NiFi CdaFilterProcessor in its pipeline

I'm going to set up this:

I accomplish this by clicking, dragging and dropping three new processors to the workspace. Here they are and their configurations:

GetFile configuration

Just the defaults...

The 60 seconds keep the processor from firing over and over again (on the same file) before I can stop the pipeline: I only need it to fire once.

Get CDA_Sample1.xml out of /tmp/nifi-sources for the test.

From GetFile, a connection to CdaFilterProcessor must be created for the success relationship (I only pass this file along if it exists):

CdaFilterProcessor configuration

From CdaFilterProcessor, a connection to PutFile must be created for the original and success relationships:

PutFile configuration

Running the test..

(Of course, you must ensure that the source file, rules list and rules are in place for the test to have any chance of running.)

Once the three processors and two connections set up, you can turn the processors on in any order, but only once two are on at the same time, with data in the pipeline, will there be processing. Logically, GetFile if the first to light up.

1. Click GetFile to select, then click the green start button in the button bar above.
2. Next, select CdaFilterProcessor and start it too. It might not start, getting "No eligible components are selected. Please select the components to be started and ensure they are no longer running." If you thought you had set everything up right, dismiss this alert and hover the mouse pointer over the yellow, warning triangle of this processor. You may see something like, "'Path to rule pathnames' is invalid. Path to rule pathnames is required." Likely, you forgot to specify this path, as shown above.
3. If nothing is running so far, did you copy CDA_Sample1.xml to /tmp/nifi-source? If not copy it. Note: It likely will not make a difference if you don't wait for the 60 seconds you set on GetFile. You can change this setting or just wait.
4. Now, once you see bytes in the CdaFilterProcessor, this means you've run something.
5. Click to select PutFile, then click the green start button.
6. Stop GetFile (before it resubmits CDA_Sample1.xml to the filter).
7. Check /tmp/filter-output for CDA_Sample1.cxml.

Monday, 7 March 2016

Have left off busy with other stuff. I just found out that the reason additionalDetails.html doesn't work is because one's expected—and this isn't what the JSON example I used to get started does—to produce a JAR of one's processor, then only wrap that in the NAR.

Without that, your trip to right-click the processor, then choosing Usage is unrewarding.

Wednesday, 9 March 2016

Attacking NiFi attributes today. You get them off a flow file. However, to add an attribute, putAttribute() returns a new or wrapped flow file. The attribute won't be there if you look at the old flow file. In this snippet, we just forget the old flow file and use the new one under the same variable.

FlowFile flowfile        = session.get();
String   attribute-value = flowfile.getAttribute( "attribute-name" );

flowfile  = session.putAttribute( flowfile, attributeName, attributeValue );

Wednesday, 23 March 2016

Remove debugging of a custom NiFi processor

I'm working on remote debugging of NiFi processors using IntelliJ IDEA.

Wednesday, 13 April 2016

An exception that is not handled by the NiFi processor is therefore handled by the framework by doing a session roll-back and administrative yield. The flowfile is put back into the queue whence it was taken before the exception. It will continue to be submitted to the processor though the processor will slow down (because yielding). Once solution is to fix the processor to catch then gate failure down a relationship.

Thursday, 14 April 2016

Got "DocGenerator Unable to document: class xyz / NullPointerException" reported in nifi-app.log

This occurs (at least) when your processor's relationships field (and the return from getRelationships()) is null.

2016-04-14 14:23:26,529 WARN [main] o.apache.nifi.documentation.DocGenerator Unable to document: class com.etretatlogiciels.nifi.processor.HeaderFooterIdentifier
java.lang.NullPointerException: null
    at org.apache.nifi.documentation.html.HtmlProcessorDocumentationWriter.writeRelationships(HtmlProcessorDocumentationWriter.java:200) ~[nifi-documentation-0.4.1.jar:0.4.1]
    at org.apache.nifi.documentation.html.HtmlProcessorDocumentationWriter.writeAdditionalBodyInfo(HtmlProcessorDocumentationWriter.java:47) ~[nifi-documentation-0.4.1.jar:0.4.1]
    at org.apache.nifi.documentation.html.HtmlDocumentationWriter.writeBody(HtmlDocumentationWriter.java:129) ~[nifi-documentation-0.4.1.jar:0.4.1]
    at org.apache.nifi.documentation.html.HtmlDocumentationWriter.write(HtmlDocumentationWriter.java:67) ~[nifi-documentation-0.4.1.jar:0.4.1]
    at org.apache.nifi.documentation.DocGenerator.document(DocGenerator.java:115) ~[nifi-documentation-0.4.1.jar:0.4.1]
    at org.apache.nifi.documentation.DocGenerator.generate(DocGenerator.java:76) ~[nifi-documentation-0.4.1.jar:0.4.1]
    at org.apache.nifi.NiFi.(NiFi.java:123) [nifi-runtime-0.4.1.jar:0.4.1]
    at org.apache.nifi.NiFi.main(NiFi.java:227) [nifi-runtime-0.4.1.jar:0.4.1]

Monday, 18 April 2016

Here's how to document multiple dynamic properties. For only one, just use @DynamicProperty alone.

@SideEffectFree
@Tags( { "SomethingProcessor" } )
@DynamicProperties( {
  @DynamicProperty(
    name = "section-N.name",
    value = "Name",
    supportsExpressionLanguage = false,
    description = "Add the name to identify a section."
        + " N is a required integer to unify the three properties especially..."
        + " Example: \"section-1.name\" : \"illness\"."
  ),
  @DynamicProperty(
    name = "section-N.pattern",
    value = "Pattern",
    supportsExpressionLanguage = false,
    description = "Add the pattern to identify a section."
        + " N is a required integer to unify the three properties especially..."
        + " Example: \"section-1.pattern\" : \"History of Illness\"."
  ),
  @DynamicProperty(
    name = "section-N.NLP",
    value = "\"on\" or \"off\"",
    supportsExpressionLanguage = false,
    description = "Add whether to turn NLP on or off."
        + " N is a required integer to unify the three properties especially..."
        + " Example: \"section-1.nlp\" : \"on\"."
  )
} )
public class SomethingProcessor extends AbstractProcessor
{
  ...

Friday and Monday, 22 & 25 April 2016

Tuesday, 26 April 2016

How to inject flow-file attributes:

TestRunner runner = TestRunners.newTestRunner( new VelocityTemplating() );

Map< String, String > flowFileAttributes = new HashMap<>();
flowFileAttributes.put( "name", "Velocity-templating Unit Test" );       // used in merging "Hello $name!"
.
.
.
runner.enqueue( DOCUMENT, flowFileAttributes );
.
.
.

If you install NiFi as a service, then wish to remove it, there are some files (these might not be the exact names) you'll need to remove:

• /etc/rc2.d/init.d/S65nifi
• /etc/rc2.d/init.d/K65nifi
• /etc/init.d/nifi

Friday, 29 April 2016

Some useful comments on examing data during flow and debugging NiFi processors.

Andy LoPresto:

NiFi is intentionally designed to allow you to make changes on a very tight cycle while interacting with live data. This separates it from a number of tools that require lengthy deployment processes to push them to sandbox/production environments and test with real data. As one of the engineers says frequently, "NiFi is like digging irrigation ditches as the water flows, rather than building out a sprinkler system in advance."

Each processor component and queue will show statistics about the data they are processing and you can see further information and history by going into the Stats dialog of the component (available through the right-click context menu on each element).

Thursday, 9 June 2016

Before explaining how to work with templates and process groups, it's necessary to define certain visual elements of the NiFi workspace. I'm just getting around to this because my focus until now has been to write and debug custom processors for others' use.

• The Components Toolbar—the eight icons on the toolbar at the left, the biggest buttons.
• The Actions Toolbar—the ten icons of the middle toolbar.
• The Management Toolbar—the eight, rightmost icons on the toolbar.

NiFi templates

Here's how to make a template in NiFi. Templates are used to reproduce work flows without having to rebuild them, from all their components and configuration, by hand. I don't know how yet to "carry" templates around between NiFi installations.

Here's how to consume an existing template in your NiFi workspace:

1. Click on the Template icon in the Components Toolbar.
2. Drag the template to the workspace.
If there are no templates, Instantiate Template will tell you.
3. Select the desired template from the modal's drop-down; hovering over one you'll see the description appear.

Process groups

You can turn a template into a process group in order to push the detail of how it's built down to make your work flow simpler at the top level.

Exporting templates for use in other NiFi installations

Importing a template

This uses the Template Management dialog too.

Monday, 13 June 2016

Some NiFi notes here on logging...

Beginning in NiFi 1.0, it can be said that it used to be that NiFi logging would be done by each processor at the INFO level to say what it's doing for each flowfile. However, this became extremely abusive in terms of logging diskspace, so the the minimum level for processors was instead set to WARN. In conf/logback.xml this can be changed back by setting the log level of org.apache.nifi.processors.

Prior to 1.0, however, one saw this:

.
.
.
<logger name="org.apache.nifi" level="INFO" />
.
.
.

Friday, 8 July 2016

Today I'm studying the NiFi ReST (NiFi API) which provides programmatic access to command and control of a NiFi instance (in real time). Specifically, I'm interested in monitoring, indeed, in obtaining an answer to questions like:

1. How many times a processor processed something?
2. How many flowfiles were processed?
3. How many flowfiles were produced?
4. Can I get this information per processor?
5. How can I tell different instances of the same processor apart?
   • ...in what's reported?
   • ...in asking about a specific processor instance?

Also, I'm interested in:

1. Getting a list of items on the NiFi canvas like processors and relationships.
2. How many times processors or a processor failed?
3. What is the profile of resource usage?
   • memory consumed
   • ?
4. What is the profile of processor latency?
   • flowfiles backed up awaiting processing
   • flowfiles in other states (?)

But first, let's break this ReST control down by service. I'm not interested in all of these today, but since I've got the information in a list, here it is:

Let me make a quick list of end-points that pique my interest and I'll try them out, make further notes, etc.

Wednesday, 13 July 2016

Thursday, 14 July 2016

Today, I followed these steps to reporting task happiness—as an experiment. I removed all log files in order to minimize the confusion and see differences in less space to look through.

Establish status quo...
1. Removed all NiFi logs and bounced NiFi, running NiFi 0.6.1.
2. Clicked on Controller Settings icon in the Management Toolbar.
3. Clicked Reporting Tasks tab.
4. Configured as reporting task leaving defaults (in particular, Run schedule 5 minutes).
5. Clicked Start (green arrow) icon.
Begin first observation...
6. Removed all NiFi logs and bounced NiFi.
7. I note that ControllerStatusReportingTask is running after bounce.

   fgrep reporting *.log yields:

   nifi-app.log:	org.apache.nifi.reporting.ganglia.StandardGangliaReporter || org.apache.nifi.nar.NarClassLoader[./work/nar/extensions/nifi-standard-nar-0.6.1.nar-unpacked]
   nifi-app.log:	org.apache.nifi.reporting.ambari.AmbariReportingTask || org.apache.nifi.nar.NarClassLoader[./work/nar/extensions/nifi-ambari-nar-0.6.1.nar-unpacked]
   nifi-user.log:2016-07-14 09:49:09,509 INFO [NiFi Web Server-25] org.apache.nifi.web.filter.RequestLogger Attempting request for (anonymous) GET http://localhost:8080/nifi-api/controller/reporting-task-types (source ip: 127.0.0.1)
   nifi-user.log:2016-07-14 09:49:09,750 INFO [NiFi Web Server-70] org.apache.nifi.web.filter.RequestLogger Attempting request (ibid)
   nifi-user.log:2016-07-14 09:49:26,596 INFO [NiFi Web Server-24] org.apache.nifi.web.filter.RequestLogger Attempting request (ibid)
   

8. I launched GetFile (Get PDF files from test fodder) and waited a few seconds for it to run and produce flowfiles.
9. I launched PutInbox (Put PDFs as in-box items) and waited a few seconds for it to run and produce flowfiles.

   Even after waiting minutes, I see no difference when running the grep. However, I launch vim on nifi-app.log and see, here and there, stuff like:

   2016-07-14 09:48:50,481 INFO [Timer-Driven Process Thread-1] c.a.n.c.C.Processors Processor Statuses:
   ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
   | Processor Name                 | Processor ID                         | Processor Type     | Run Status |             Flow Files In |            Flow Files Out |         Bytes Read |      Bytes Written |   Tasks |                    Proc Time |
   ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
   | Generate XML from advance dire | c396af51-2221-4bfd-ba31-5f4b5913b910 | VelocityTemplating |    Stopped | 0 / 0 bytes (+0/+0 bytes) | 0 / 0 bytes (+0/+0 bytes) | 0 bytes (+0 bytes) | 0 bytes (+0 bytes) |  0 (+0) | 00:00:00.000 (+00:00:00.000) |
   | Get PDFs from inbox            | 9929acf6-aac7-46d9-a0dd-f239d66e1594 | GetInbox           |    Stopped | 0 / 0 bytes (+0/+0 bytes) | 0 / 0 bytes (+0/+0 bytes) | 0 bytes (+0 bytes) | 0 bytes (+0 bytes) |  0 (+0) | 00:00:00.000 (+00:00:00.000) |
   | Put PDFs as in-box items       | 5f28baeb-b24c-4c3f-b6d0-c3284553d6ac | PutInbox           |    Stopped | 0 / 0 bytes (+0/+0 bytes) | 0 / 0 bytes (+0/+0 bytes) | 0 bytes (+0 bytes) | 0 bytes (+0 bytes) |  0 (+0) | 00:00:00.000 (+00:00:00.000) |
   | Put generated XMLs to output f | 1af4a618-ca9b-4697-81b5-3ba1bbfaba6b | PutFile            |    Stopped | 0 / 0 bytes (+0/+0 bytes) | 0 / 0 bytes (+0/+0 bytes) | 0 bytes (+0 bytes) | 0 bytes (+0 bytes) |  0 (+0) | 00:00:00.000 (+00:00:00.000) |
   | Get PDF files from test fodder | 4435c701-0719-4231-8867-671db08dd813 | GetFile            |    Stopped | 0 / 0 bytes (+0/+0 bytes) | 0 / 0 bytes (+0/+0 bytes) | 0 bytes (+0 bytes) | 0 bytes (+0 bytes) |  0 (+0) | 00:00:00.000 (+00:00:00.000) |
   ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
   2016-07-14 09:48:50,483 INFO [Timer-Driven Process Thread-1] c.a.n.c.C.Connections Connection Statuses:
   ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
   | Connection ID                        | Source                         | Connection Name | Destination                    |              Flow Files In |             Flow Files Out |             FlowFiles Queued |
   ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
   | 53e4a0e9-ce20-4e4d-84f3-d12504edeccc | Get PDF files from test fodder | success         | Put PDFs as in-box items       |  0 / 0 bytes (+0/+0 bytes) |  0 / 0 bytes (+0/+0 bytes) |  9 / 75.82 KB (+9/+75.82 KB) |
   | f3a032d6-6bd1-4cc5-ab5e-8923e131c222 | Get PDFs from inbox            | success         | Generate XML from advance dire |  0 / 0 bytes (+0/+0 bytes) |  0 / 0 bytes (+0/+0 bytes) |    0 / 0 bytes (+0/+0 bytes) |
   | e5ba7858-8301-4fd0-885e-6fb63b336e53 | Generate XML from advance dire | success         | Put generated XMLs to output f |  0 / 0 bytes (+0/+0 bytes) |  0 / 0 bytes (+0/+0 bytes) |    0 / 0 bytes (+0/+0 bytes) |
   ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
   

   (I would figure out a little later that this is the start-up view before anthing's happened.)

Begin attempting to redirect reporting task observation...
10. I inserted this near the bottom of conf/logback.xml:

    <!-- Logger to redirect reporting task output to a different file. -->
    <logger name="org.apache.nifi.controller.ControllerStatusReportingTask" level="INFO" additivity="false">
      <appender-ref ref="STATUS_FILE" />
    </logger>
    

11. ...and I inserted this near the top of conf/logback.xml, just after the definition of BOOTSTRAP_FILE rolling appender:

    <!-- rolling appender for controller-status reporting task log. -->
    <appender name="STATUS_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
      <file>logs/nifi-status.log</file>
      <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
        <fileNamePattern>./logs/nifi-status_%d.log</fileNamePattern>
        <maxHistory>30</maxHistory>
      </rollingPolicy>
      <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
        <pattern>%date %level [%thread] %logger{40} %msg%n</pattern>
      </encoder>
    </appender>
    

12. Removed all NiFi logs and bounced NiFi.
13. Looked at logs:

    ~/dev/nifi/logs $ ll
    total 76
    drwxr-xr-x  2 russ russ  4096 Jul 14 10:28 .
    drwxr-xr-x 13 russ russ  4096 Jul 14 09:37 ..
    -rw-r--r--  1 russ russ 65016 Jul 14 10:29 nifi-app.log
    -rw-r--r--  1 russ russ  3069 Jul 14 10:28 nifi-bootstrap.log
    -rw-r--r--  1 russ russ     0 Jul 14 10:28 nifi-status.log
    -rw-r--r--  1 russ russ     0 Jul 14 10:28 nifi-user.log
    

14. I launched GetFile (Get PDF files from test fodder) and waited a few seconds for it to run and produce flowfiles.

    

15. I launched PutInbox (Put PDFs as in-box items) and waited a few seconds for it to run and produce flowfiles.

    

16. I never see nifi-status.log grow in size. I launch vim on nifi-app.log and see, just as before (but this time I'm scraping the stuff showing work accomplished in bold):

    
    2016-07-14 10:34:06,643 INFO [Timer-Driven Process Thread-1] c.a.n.c.C.Processors Processor Statuses:
    -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
    | Processor Name                 | Processor ID                         | Processor Type     | Run Status |               Flow Files In |              Flow Files Out |           Bytes Read |        Bytes Written |  Tasks |                    Proc Time |
    -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
    | Get PDF files from test fodder | 4435c701-0719-4231-8867-671db08dd813 | GetFile            |    Stopped |   0 / 0 bytes (+0/+0 bytes) | 6 / 50.54 KB (+6/+50.54 KB) | 50.54 KB (+50.54 KB) | 50.54 KB (+50.54 KB) | 2 (+2) | 00:00:00.056 (+00:00:00.056) |
    | Put PDFs as in-box items       | 5f28baeb-b24c-4c3f-b6d0-c3284553d6ac | PutInbox           |    Stopped | 6 / 50.54 KB (+6/+50.54 KB) |   0 / 0 bytes (+0/+0 bytes) | 50.54 KB (+50.54 KB) |   0 bytes (+0 bytes) | 1 (+1) | 00:00:00.014 (+00:00:00.014) |
    | Generate XML from advance dire | c396af51-2221-4bfd-ba31-5f4b5913b910 | VelocityTemplating |    Stopped |   0 / 0 bytes (+0/+0 bytes) |   0 / 0 bytes (+0/+0 bytes) |   0 bytes (+0 bytes) |   0 bytes (+0 bytes) | 0 (+0) | 00:00:00.000 (+00:00:00.000) |
    | Get PDFs from inbox            | 9929acf6-aac7-46d9-a0dd-f239d66e1594 | GetInbox           |    Stopped |   0 / 0 bytes (+0/+0 bytes) |   0 / 0 bytes (+0/+0 bytes) |   0 bytes (+0 bytes) |   0 bytes (+0 bytes) | 0 (+0) | 00:00:00.000 (+00:00:00.000) |
    | Put generated XMLs to output f | 1af4a618-ca9b-4697-81b5-3ba1bbfaba6b | PutFile            |    Stopped |   0 / 0 bytes (+0/+0 bytes) |   0 / 0 bytes (+0/+0 bytes) |   0 bytes (+0 bytes) |   0 bytes (+0 bytes) | 0 (+0) | 00:00:00.000 (+00:00:00.000) |
    -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
    2016-07-14 10:34:06,644 INFO [Timer-Driven Process Thread-1] c.a.n.c.C.Connections Connection Statuses:
    --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
    | Connection ID                        | Source                         | Connection Name | Destination                    |               Flow Files In |              Flow Files Out |          FlowFiles Queued |
    --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
    | f3a032d6-6bd1-4cc5-ab5e-8923e131c222 | Get PDFs from inbox            | success         | Generate XML from advance dire |   0 / 0 bytes (+0/+0 bytes) |   0 / 0 bytes (+0/+0 bytes) | 0 / 0 bytes (+0/+0 bytes) |
    | e5ba7858-8301-4fd0-885e-6fb63b336e53 | Generate XML from advance dire | success         | Put generated XMLs to output f |   0 / 0 bytes (+0/+0 bytes) |   0 / 0 bytes (+0/+0 bytes) | 0 / 0 bytes (+0/+0 bytes) |
    | 53e4a0e9-ce20-4e4d-84f3-d12504edeccc | Get PDF files from test fodder | success         | Put PDFs as in-box items       | 6 / 50.54 KB (+6/+50.54 KB) | 6 / 50.54 KB (+6/+50.54 KB) | 0 / 0 bytes (+0/+0 bytes) |
    --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
    

So, this was a partial failure...

What failed is getting the reported status into nifi-status.log. What succeeded is, just as was working before, getting controller status.

Tuesday, 23 August 2016

Apache NiFi expression language notes

I need to take time out to grok this.

Friday, 26 August 2016

Thursday, 8 September 2016

Here's one for your gee-whiz collection: if you do

public void onTrigger( final ProcessContext context, final ProcessSession session ) throws ProcessException
{
  FlowFile flowfile = session.get();
  .
  .
  .
  someMethod( session );

...NiFi has removed the flowfile from the session's processorQueue and a subsequent call:

void someMethod( final ProcessSession session )
{
  FlowFile flowfile = session.get();

...will produce a null flowfile. This is an interesting side-effect.

Wednesday, 14 September 2016

Missing flowfiles...

In what situation would onTrigger() get called if there is no flowfile?

This can happen for a few reasons. Because the processor has...

1. ...no incoming connections (it's a source processor),
2. ...@ScheduleWhenEmpty annotation,
3. ...more than one concurrent task,
4. (others?)

The main reason is the third. If you have multiple concurrent tasks, Thread 1 can determine that there is 1 flowfile queued. Thread 2 then determines that there is one flowfile queued. Both threads call onTrigger(), thread 1 gets the flowfile, and thread 2 gets nothing (null).

Backing up NiFi flows, using templates, etc....

Is there a best practice for keeping a backup of data flows, assuming using flow.xml.gz and configuration files, it's possible to restore NiFi in case of any failure? Is it possible simply to "drop" this file into a new NiFi installation or should templates be used in development that get imported into production?

It's a good idea to back up flow.xml.gz and configuration files. But distinguish between using these backups to recreate an equivalent new NiFi installation and attempting to reset the state of the existing one. The difference is the live data in the flow, in the provenance repository, in state variables, etc.: Restoring a flow definition that doesn't match (or no longer matches) your content and provenance data may have unexpected results for you and for systems connecting with the NiFi installation. NiFi tries hard to handle these changes smoothly, but it isn't a magic time machine.

Deploying flow.xml.gz can work, especially when deployed with configuration files that reference IDs in the flow (like authorizations.xml) or the nifi.sensitive.props.key setting, and others. But if you overwrite a running flow, you will still have data migration problems.

Templates are the current recommended best practice for deployment. They provide:

1. concise packaging for deployment,
2. separation between site-specific configuration like authorizations from the flow logic,
3. workflow that allows, encourages and forces the administrator to address migration from the existing flow to incorporate the new template.

Is there a way to predeploy templates to the template directory and have NiFi can recognize them?

Templates have been polished quite a bit to be more deterministic and source-control friendly so they are an ideal artifact to be versioned and kept in source-control repository.

When NiFi starts up, it looks in conf/templates (by default—this directory can be changed in nifi.properties). It looks for any file that has a suffix of .template or .xml so be sure the files are named properly. If starting a cluster, ensure all nodes in the cluster have these same templates or the latter will not show up.

Friday, 23 September 2016

Here's a small, complete example of a processor that translates attributes names or removes attributes. I found UpdateAttributes to be overly complicated and hard to use.

NiFiStandardAttributes.java:

package com.etretatlogiciels.nifi.attributes;

import java.util.ArrayList;
import java.util.List;

public class NiFiStandardAttributes
{
  public static final List< String > FLOWFILE_ATTRIBUTES;

  static
  {
    FLOWFILE_ATTRIBUTES = new ArrayList<>( 5 );
    FLOWFILE_ATTRIBUTES.add( "filename" );
    FLOWFILE_ATTRIBUTES.add( "uuid" );
    FLOWFILE_ATTRIBUTES.add( "path" );
  }

  public static final int FLOWFILE_ATTRIBUTE_COUNT = FLOWFILE_ATTRIBUTES.size();
}

TranslateAttributeNamesTest.java:

package com.etretatlogiciels.nifi.processor;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;

import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertNotNull;

import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;

import com.etretatlogiciels.nifi.attributes.NiFiStandardAttributes;
import com.etretatlogiciels.testing.TestUtilities;

/**
 * @author Russell Bateman
 * @since September 2016
 */
public class TranslateAttributeNamesTest
{
  // @formatter:off
  @Rule   public TestName name = new TestName();
  @Before public void setUp() throws Exception { TestUtilities.setUp( name ); }
  @After  public void tearDown() throws Exception { }

  private static final boolean VERBOSE = true;//TestUtilities.VERBOSE;

  private static final String TRANSLATE_JSON ="{"
                                            + "    \"inner-MIME\" : \"mime-type\","
                                            + "    \"lose-this\" : \"\","
                                            + "    \"lose-this-too\" : null"
                                            + "}";

  @Test
  public void testBasic()
  {
    TestRunner runner = TestRunners.newTestRunner( new TranslateAttributeNames() );

    final int         ONE      = 1;
    final String      CONTENTS = "This is a test whose flowfile contents are insignificant!";
    final InputStream MESSAGE  = new ByteArrayInputStream( CONTENTS.getBytes() );

    runner.setProperty( TranslateAttributeNames.TRANSLATIONS, TRANSLATE_JSON );

    Map< String, String > flowFileAttributes = new HashMap<>();
    flowFileAttributes.put( "testname",   "X12 Message Router Unit Test" );
    flowFileAttributes.put( "inner-MIME", "application/pdf" );
    flowFileAttributes.put( "lose-this", "" );
    flowFileAttributes.put( "lose-this-too", null );
    flowFileAttributes.put( "dont-touch-this-one", "Don't touch this attribute!" );

    if( VERBOSE )
    {
      System.out.println( "Attributes on the original flowfile:" );
      Map< String, String > sortedAttributes = new TreeMap<>( flowFileAttributes );
      for( Map.Entry< String, String > attribute : sortedAttributes.entrySet() )
        System.out.println( "  " + attribute.getKey() + " = " + attribute.getValue() );
    }

    runner.setValidateExpressionUsage( false );
    runner.enqueue( MESSAGE, flowFileAttributes );

    runner.run( ONE );
    runner.assertQueueEmpty();
    List< MockFlowFile > results = runner.getFlowFilesForRelationship( TranslateAttributeNames.SUCCESS );
    MockFlowFile         result  = results.get( 0 );
    String               actual  = new String( runner.getContentAsByteArray( result ) );

    if( VERBOSE )
    {
      System.out.println( "Surviving flowfile contents:" );
      System.out.println( "  " + actual );
      System.out.println( "Attributes surviving on the flowfile:" );
      Map< String, String > sortedAttributes = new TreeMap<>( result.getAttributes() );
      for( Map.Entry< String, String > attribute : sortedAttributes.entrySet() )
      {
        String key = attribute.getKey();

        if( !NiFiStandardAttributes.FLOWFILE_ATTRIBUTES.contains( key ) )
          System.out.println( "  " + key + " = " + attribute.getValue() );
      }
    }

    assertTrue( "Flowfile contents changed", actual.equals( CONTENTS ) );

    String mimeType = result.getAttribute( "mime-type" );
    assertNotNull( "Attribute \"inner-MIME\" was not renamed", mimeType );
    assertTrue( "Attribute \"inner-MIME\" was not renamed", mimeType.equals( "application/pdf" ) );

    String dontTouchThisOne = result.getAttribute( "dont-touch-this-one" );
    assertNotNull( "Attribute \"dont-touch-this-one\" did not survive", result.getAttribute( "dont-touch-this-one" ) );
    assertTrue( "Attribute \"dont-touch-this-one\" did not survive untouched",
                 dontTouchThisOne.equals( "Don't touch this attribute!" ) );
    assertNull( "Attribute \"lose-this\" was not dropped", result.getAttribute( "lose-this" ) );
    assertNull( "Attribute \"lose-this-too\" was not dropped", result.getAttribute( "lose-this-too" ) );
  }
}

TranslateAttributeNamesTest.java:

package com.etretatlogiciels.nifi.processor;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;

import org.apache.nifi.annotation.behavior.SideEffectFree;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.Validator;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;

import com.etretatlogiciels.nifi.rel.StandardRelationships;
import com.etretatlogiciels.utilities.StringUtilities;

/**
 * Translate attribute names or even drop attributes altogether.
 *
 * {
 *   "attachment-filename"  : "attachment",
 *   "disposition-encoding" : "encoding",
 *   "disposition-type"     : null,
 *   "inner-MIME"           : "mime_type",
 *   "inner-boundary"       : "Donald Duck
 * }
 *
 * @author Russell Bateman
 * @since September 2016
 */
@SideEffectFree
@Tags( { "translateattributenames" } )
@CapabilityDescription( "Translate the names of attributes to new ones. The flowfile itself remains untouched." )
public class TranslateAttributeNames extends AbstractProcessor
{
  @Override
  public void onTrigger( final ProcessContext context, final ProcessSession session ) throws ProcessException
  {
    FlowFile flowfile = session.get();

    Map< String, String > translations = new Gson().fromJson( context.getProperty( TRANSLATIONS ).getValue(),
                                  new TypeToken< Map< String, String > >(){ }.getType() );

    // determine which attributes we'll be pushing onto the new flowfile...
    Map< String, String > candidates = flowfile.getAttributes();
    Map< String, String > attributes = new HashMap<>();

    for( Map.Entry< String, String > attribute : candidates.entrySet() )
    {
      String key = attribute.getKey();

      // do nothing because we're leaving the attribute as-is...
      if( !translations.containsKey( key ) )
        continue;

      // remove this flowfile attribute because we're changing its name or dropping it...
      flowfile = session.removeAttribute( flowfile, key );

      String newKey = translations.get( key );

      // we're dropping this attribute altogether...
      if( StringUtilities.isEmpty( newKey ) || newKey.equals( "null" ) )
        continue;

      // here's the attribute under its new name...
      attributes.put( translations.get( key ), attribute.getValue() );
    }

    flowfile = session.putAllAttributes( flowfile, attributes );
    session.transfer( flowfile, SUCCESS );
  }

  private List< PropertyDescriptor > properties    = new ArrayList<>();
  private Set< Relationship >        relationships = new HashSet<>();

  @Override
  public void init( final ProcessorInitializationContext context )
  {
    properties.add( TRANSLATIONS );
    properties = Collections.unmodifiableList( properties );
    relationships.add( SUCCESS );
    relationships = Collections.unmodifiableSet( relationships );
  }

  @Override public List< PropertyDescriptor> getSupportedPropertyDescriptors() { return properties; }
  @Override public Set< Relationship >       getRelationships()                { return relationships; }

  //<editor-fold desc="Processor properties and relationships">
  public static final PropertyDescriptor TRANSLATIONS = new PropertyDescriptor.Builder()
      .name( "Translations" )
      .description( "JSON detailing the name changes as key-value pairs where the value is the new name. Any value "
          + "that's null or <"<" will result in the attribute being dropped (not transmitted on the flowfile)." )
      .required( false )
      .addValidator( Validator.VALID )
      .build();

  public static final Relationship SUCCESS = StandardRelationships.SUCCESS;
  //</editor-fold>
}

Wednesday, 28 September 2016

I had a question about cloning the input stream:

  // under some condition, look into the flowfile contents to see if something's there...
  session.read( flowfile, new InputStreamCallback()
  {
    @Override
    public void process( InputStream( in ) throws IOException
    {
      // read from in...
    }
  } );


  // then, later (and always) consume it (so, sometimes a second time):

  session.read( flowfile, new InputStreamCallback()
  {
    @Override
    public void process( InputStream( in ) throws IOException
    {
    // read from in (maybe again)...
    }
  } );

Wouldn't the second time be problematic since the stream contents have already been read?

I learned from Mark Payne that each time session.read() is called, I'm getting a new InputStream that starts at the beginning of the flowfile. How convenient!

Thursday, 29 September 2016

Some observations on using GetFile:

1. Configure Scheduling → Run schedule to 10000 sec. This makes it so that this processor will only run once, then wait 10.000 seconds before attempting to run again.
2. Running "once" means that if the File filter template is adequate, it will likely induct all or many of the files on the Input directory no matter how quickly you can reach for the stop button.
3. So, if you only want one file at a time, only put one file in the Input subdirectory that matches the File filter template.

Friday, 30 September 2016

I don't know why, after all these months, I'm still making mistakes like this. In two of my processors, which work when unit-tested, I keep getting:

Could not read from StandardFlowFileRecord

I tried a new way because likely the InputStream I'm caching is getting closed before I can use it again and this doesn't show up in the unit-test environment:

    FlowFile flowfile = session.get();

    flowfile = session.write( flowfile, new StreamCallback()
    {
      @Override
      public void process( InputStream inputStream, OutputStream outputStream ) throws IOException
      {
        Base64Unroller unroller = new Base64Unroller( inputStream );
        unroller.unroll();

        InputStream in = unroller.getBase64Data();
        int         c;

        for( ; ; )
        {
          c = in.read();
          if( c == EOS || ( c == '\n' || c == '\r' ) )
            break;
          outputStream.write( c );
        }
      }
    } );

    session.transfer( flowfile, SUCCESS );

Flush with this better understanding of input- and output streams in NiFi, I'm going to see if this doesn't answer also the problem I have with Base64Decode.

It does solve my problem there: this is the answer to cases where I need both to read the input stream and write to an output stream.

Addendum: Processor patterns

I'm more interested here in detailing coding clauses I often write, but I don't write processors of all these sorts. Hilighted are lines important to consider.

• Data ingress—data into NiFi (read alone):

  final FlowFile flowfile = session.get();
  session.read( flowfile, new InputStreamCallback()
  {
    @Override
    public void process( InputStream in ) throws IOException
    {
      /* do stuff on the input stream */
  
  

• Data egress—data out of NiFi (write alone):

  FlowFile flowfile = session.create();
  flowfile = session.write( flowfile, new OutputStreamCallback()
  {
    @Override public void process( OutputStream out ) throws IOException
    {
      out.write( /* the input from somewhere other than the flowfile */ );
    }
  });
  
  

• Route based on content (one-to-one)—...
• Route based on content (one-to-many)—...
• Route streams-based on content (one-to-many)—...
• Route based on attributes—doesn't use ProcessSession's read() method because doesn't read flowfile content.
• Split content (one-to-many)—implies a list of flowfiles to create and configuration as to where to split content.
• Enrich and modify content—reading from a flowfile and writing to the next version of the flowfile's content:

  final FlowFile flowfile = session.get();
  flowfile = session.write( flowfile, new StreamCallback()
  {
    @Override public void process( InputStream inputStream, OutputStream outputStream )
        throws IOException
    {
      InputStreamReader  input  = new InputStreamReader (  inputStream, "UTF-8" );
      OutputStreamReader output = new OutputStreamReader( outputStream, "UTF-8" );
  		// read from input stream; write to output stream...
    }
  });
  
  

Monday, 10 October 2016

A list of repositories and what they do.

• flowfile repository, data records, i.e.: the metadata of the flowfiles
• content repository, the content of the flowfiles themselves
• provenance repository, record of whence the flowfile comes, what's happened to it along the way
• database repository, various metadata/configuration for NiFi including user access

Tuesday, 11 October 2016

I asked a question in the [email protected] forum.

"If I create flows for integration testing of multiple components (say, processors), where can I find the underlying files I can pick up, store and deploy to a fresh server with a fresh NiFi, then launch them on data (that I also deploy)?"

"I assumed that once I 'paint' my canvas with processors and relationships, it's harvestable in some form, in some .xml file somewhere, and easily transported to, then dropped into, a new NiFi installation on some server or host for running it. I assumed my testing would be thus configured, then kicked off somehow."

Andy LoPresto answered, "The 'flow' (the processors on the canvas and the relationships between them) is stored on disk as ${NIFI_HOME}/conf/flow.xml.gz [1]. You can also export selections from the flow as templates [2], along with a name and description, which can then be saved as XML files and loaded into a new instance of NiFi with (non-sensitive) configuration values and properties maintained. You may also be interested in the ongoing work to support the 'Software Development Lifecycle (SDLC)' or 'Development to Production Pipeline (D2P)' efforts to allow users to develop flows in testbed environments and 'promote' them through code control/repositories to QE/QA and then Production environments [3].

"If you are attempting to perform automated testing of flows or flow segments, you can do this with the test runner [4]. I would encourage you to store the flow in a template or full flow file (flow.xml.gz, not a 'flowfile') and load that during the test rather than programmatically instantiating and configuring each component and relationship in the test code, but you are able to do both. Also, be aware that the TestRunner is just that, and approximates/mocks some elements of the NiFi environment, so you should understand what decisions are being made behind the scenes to ensure that what is covered by the test is exactly what you expect."

Wednesday, 12 October 2016

It's possible to annotate the NiFi canvas with labels, clicking and dragging a label to the the canvas and configuring it. These labels land behind anything you constructively design on the canvas (processors, relationships, funnels, etc.).

I have a hard time remembering how to view flowfile content in a queue, here are the steps:

1. Right-click the queue,
2. Choose List queue,
3. At the extreme left, click the (i) button, which is "View Details," of the flowfile that interests you,
4. In the Details tab, click the View button,
5. You have the option to view it as a) original (text) format, b) formatted (see content type at upper right: whether this choice makes a difference or not depends on MIME type) or c) hex dump (very useful for binary or even "confusing" text—looking for carriage returns vs. newlines, control characters, etc.).

Tuesday, 1 November 2016

On dynamic properties in NiFi...

Wednesday, 2 November 2016

Every once in a while, I'm puzzled and rediscover that I've left stuff out making me scratch my head.

If you get this error in your log or debugger...

java.lang.AssertionError: Processor has 1 validation failures:
property-name validated against value is invalid because property-name is not a supported property

..., you've likely left these methods out of your processor's implementation:

@Override public List< PropertyDescriptor > getSupportedPropertyDescriptors() { return properties; }
@Override public Set< Relationship >        getRelationships()                { return relationships; }

Here's how to get NiFi back up once you've removed a custom processor (yes, you lose your entire canvas):

~/dev/nifi/nifi-0.7.1/conf $ ll
total 72
drwxr-xr-x.  3 russ russ 4096 Nov  2 14:45 .
drwxr-xr-x. 14 russ russ 4096 Oct 21 09:07 ..
-rw-r--r--.  1 russ russ 2129 Oct 13 17:54 authority-providers.xml
-rw-r--r--.  1 russ russ 2530 Oct 13 17:54 authorized-users.xml
-rw-r--r--.  1 russ russ 3241 Oct 13 17:54 bootstrap.conf
-rw-r--r--.  1 russ russ 2119 Oct 13 17:15 bootstrap-notification-services.xml
-rw-r--r--.  1 russ russ 1317 Nov  2 14:45  flow.xml.gz
-rw-r--r--.  1 russ russ 7454 Oct 13 17:54 logback.xml
-rw-r--r--.  1 russ russ 6089 Oct 13 17:54 login-identity-providers.xml
-rw-r--r--.  1 russ russ 8571 Oct 13 17:54 nifi.properties
-rw-r--r--.  1 russ russ 4603 Oct 13 17:54 state-management.xml
drwxr-xr-x.  2 russ russ 4096 Oct 21 09:07 templates
-rw-r--r--.  1 russ russ 1437 Oct 13 17:15 zookeeper.properties
~/dev/nifi/nifi-0.7.1/conf $ rm flow.xml.gz

Friday, 4 November 2016

(I can be embarrassingly stupid sometimes...)

You've configured nifi.properties to use port 9091.

nifi.web.http.port=9091

...but can't get NiFi to launch. The error from logs/nifi-bootstrap.log is:

~/dev/nifi/nifi-1.0.0/logs $ cat nifi-bootstrap.log
2016-11-04 11:32:32,217 INFO [main] o.a.n.b.NotificationServiceManager Successfully loaded the following 0 services: []
2016-11-04 11:32:32,221 INFO [main] org.apache.nifi.bootstrap.RunNiFi Registered no Notification Services for Notification Type NIFI_STARTED
2016-11-04 11:32:32,221 INFO [main] org.apache.nifi.bootstrap.RunNiFi Registered no Notification Services for Notification Type NIFI_STOPPED
2016-11-04 11:32:32,221 INFO [main] org.apache.nifi.bootstrap.RunNiFi Registered no Notification Services for Notification Type NIFI_DIED
2016-11-04 11:32:32,222 INFO [main] org.apache.nifi.bootstrap.Command Apache NiFi is not currently running
2016-11-04 11:32:35,684 INFO [main] o.a.n.b.NotificationServiceManager Successfully loaded the following 0 services: []
2016-11-04 11:32:35,687 INFO [main] org.apache.nifi.bootstrap.RunNiFi Registered no Notification Services for Notification Type NIFI_STARTED
2016-11-04 11:32:35,688 INFO [main] org.apache.nifi.bootstrap.RunNiFi Registered no Notification Services for Notification Type NIFI_STOPPED
2016-11-04 11:32:35,688 INFO [main] org.apache.nifi.bootstrap.RunNiFi Registered no Notification Services for Notification Type NIFI_DIED
2016-11-04 11:32:35,695 INFO [main] org.apache.nifi.bootstrap.Command Starting Apache NiFi...
2016-11-04 11:32:35,695 INFO [main] org.apache.nifi.bootstrap.Command Working Directory: /home/russ/dev/nifi/nifi-1.0.0
2016-11-04 11:32:35,696 INFO [main] org.apache.nifi.bootstrap.Command Command: /home/russ/dev/jdk1.8.0_25/bin/java -classpath /home/russ/dev/nifi/nifi-1.0.0/./conf:/home/russ/dev/nifi/nifi-1.0.0/./lib/nifi-properties-loader-1.0.0.jar:/home/russ/dev/nifi/nifi-1.0.0/./lib/jcl-over-slf4j-1.7.12.jar:/home/russ/dev/nifi/nifi-1.0.0/./lib/nifi-api-1.0.0.jar:/home/russ/dev/nifi/nifi-1.0.0/./lib/logback-core-1.1.3.jar:/home/russ/dev/nifi/nifi-1.0.0/./lib/logback-classic-1.1.3.jar:/home/russ/dev/nifi/nifi-1.0.0/./lib/nifi-framework-api-1.0.0.jar:/home/russ/dev/nifi/nifi-1.0.0/./lib/jul-to-slf4j-1.7.12.jar:/home/russ/dev/nifi/nifi-1.0.0/./lib/nifi-documentation-1.0.0.jar:/home/russ/dev/nifi/nifi-1.0.0/./lib/nifi-runtime-1.0.0.jar:/home/russ/dev/nifi/nifi-1.0.0/./lib/slf4j-api-1.7.12.jar:/home/russ/dev/nifi/nifi-1.0.0/./lib/nifi-properties-1.0.0.jar:/home/russ/dev/nifi/nifi-1.0.0/./lib/bcprov-jdk15on-1.54.jar:/home/russ/dev/nifi/nifi-1.0.0/./lib/log4j-over-slf 4j-1.7.12.jar:/home/russ/dev/nifi/nifi-1.0.0/./lib/commons-lang3-3.4.jar:/home/russ/dev/nifi/nifi-1.0.0/./lib/nifi-nar-utils-1.0.0.jar -Dorg.apache.jasper.compiler.disablejsr199=true -Xmx512m -Xms512m -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=8000 -Dsun.net.http.allowRestrictedHeaders=true -Djava.net.preferIPv4Stack=true -Djava.awt.headless=true -XX:+UseG1GC -Djava.protocol.handler.pkgs=sun.net.www.protocol -Dnifi.properties.file.path=/home/russ/dev/nifi/nifi-1.0.0/./conf/nifi.properties -Dnifi.bootstrap.listen.port=33110 -Dapp=NiFi -Dorg.apache.nifi.bootstrap.config.log.dir=/home/russ/dev/nifi/nifi-1.0.0/logs org.apache.nifi.NiFi
2016-11-04 11:32:35,747 ERROR [NiFi logging handler] org.apache.nifi.StdErr ERROR: transport error 202: bind failed: Address already in use
2016-11-04 11:32:35,747 ERROR [NiFi logging handler] org.apache.nifi.StdErr ERROR: JDWP Transport dt_socket failed to initialize, TRANSPORT_INIT(510)
2016-11-04 11:32:35,747 ERROR [NiFi logging handler] org.apache.nifi.StdErr JDWP exit error AGENT_ERROR_TRANSPORT_INIT(197): No transports initialized [debugInit.c:750]
2016-11-04 11:32:35,747 INFO [NiFi logging handler] org.apache.nifi.StdOut FATAL ERROR in native method: JDWP No transports initialized, jvmtiError=AGENT_ERROR_TRANSPORT_INIT(197)
2016-11-04 11:32:36,739 INFO [main] org.apache.nifi.bootstrap.RunNiFi NiFi never started. Will not restart NiFi

And yet,

# lsof -i | grep 9091

...produces nothing for port 9091 (nor does netstat -lptu). So, port 9091 is not in use and yet the error leads one to believe that this is the problem.

It looks like remote debugging has been configured, which specifies a port that's obviously in use (8000). That happens in bootstrap.conf:

# Enable Remote Debugging
java.arg.debug=-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=8000

Thursday, 17 November 2016

Still working on identifying how to create and store away in version control a process group.

Also looking into how we can hot-swap new or updated processors. Later, I'll get into an Agile write-up everyone agreed I'd do in the Core Team meeting yesterday afternoon.

We could set up a cluster for NiFi because rolling restarts are possible with individual instances in a cluster, that is, if I understand correctly, it's possible to take down one instance and update it, i.e.: restart it, without affecting the other though I wonder still right now if there's a way to throttle or somehow pause the instance so that it stops handling anything so that bouncing it has no ramifications like reprocessing jobs, etc.

To be noted too that the new/updated software isn't available until all nodes of the cluster have been updated and restarted.

When NiFi first starts up, the following directories (and files) are created:

• content_repository
• database_repository
• flowfile_repository
• provenance_repository
• work directory
• logs directory
• Within the conf subdirectory, ...
  • the flow.xml.gz file,
  • ...and the templates directory are created
  • authority-providers.xml, providers to use when running securely
  • authorized-users.xml, users
  • bootstrap.conf, for launching NiFi, JVM, etc.
  • bootstrap-notification-services.xml, notification of life-cycle events
  • logback.xml (like log4j.properties)
  • login-identity-providers.xml, log-in identity providers
  • nifi.properties, core properties
  • state-management.xml, for clustering
  • zookeeper.properties, for clustering

flow.xml.gz

Here's a flow to look at:

Unzipped (from flow.xml.gz), this flow looks like this:

<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<flowController>
  <maxTimerDrivenThreadCount>10</maxTimerDrivenThreadCount>
  <maxEventDrivenThreadCount>5</maxEventDrivenThreadCount>
  <rootGroup>
    <id>9a3b3f31-d3fa-4e18-853f-f49510f48c2a</id>
    <name>NiFi Flow</name>
    <position x="0.0" y="0.0"/>
    <comment/>
    <processor>
      <id>0c689901-1e21-4a89-af21-04ec13f1ca03</id>
      <name>CreateFlowFile</name>
      <position x="-27.0" y="-12.0"/>
      <styles/>
      <comment/>
      <class>com.etretatlogiciels.nifi.processor.CreateFlowFile</class>
      <maxConcurrentTasks>1</maxConcurrentTasks>
      <schedulingPeriod>0 sec</schedulingPeriod>
      <penalizationPeriod>30 sec</penalizationPeriod>
      <yieldPeriod>1 sec</yieldPeriod>
      <bulletinLevel>WARN</bulletinLevel>
      <lossTolerant>false</lossTolerant>
      <scheduledState>STOPPED</scheduledState>
      <schedulingStrategy>TIMER_DRIVEN</schedulingStrategy>
      <runDurationNanos>0</runDurationNanos>
      <property>
        <name>Content</name>
        <value>This is a test of the Emergency Broadcast System.</value>
      </property>
    </processor>
    <processor>
      <id>c3daf696-11e7-4684-9e84-9f1aaedf15a6</id>
      <name>RetryOperation</name>
      <position x="281.0" y="86.0"/>
      <styles/>
      <comment/>
      <class>com.etretatlogiciels.nifi.processor.RetryOperation</class>
      <maxConcurrentTasks>1</maxConcurrentTasks>
      <schedulingPeriod>0 sec</schedulingPeriod>
      <penalizationPeriod>30 sec</penalizationPeriod>
      <yieldPeriod>1 sec</yieldPeriod>
      <bulletinLevel>WARN</bulletinLevel>
      <lossTolerant>false</lossTolerant>
      <scheduledState>STOPPED</scheduledState>
      <schedulingStrategy>TIMER_DRIVEN</schedulingStrategy>
      <runDurationNanos>0</runDurationNanos>
      <property>
        <name>Stop retrying operation when</name>
      </property>
      <property>
        <name>Flowfile attribute name</name>
      </property>
      <property>
        <name>Initial counter value</name>
        <value>3</value>
      </property>
      <property>
        <name>Retry cache-table TTL</name>
        <value>10</value>
      </property>
    </processor>
    <processor>
      <id>e1d7c27f-e22f-480d-9d8d-591accbb7180</id>
      <name>NoOperation</name>
      <position x="639.0" y="272.0"/>
      <styles/>
      <comment/>
      <class>com.etretatlogiciels.nifi.processor.NoOperation</class>
      <maxConcurrentTasks>1</maxConcurrentTasks>
      <schedulingPeriod>0 sec</schedulingPeriod>
      <penalizationPeriod>30 sec</penalizationPeriod>
      <yieldPeriod>1 sec</yieldPeriod>
      <bulletinLevel>WARN</bulletinLevel>
      <lossTolerant>false</lossTolerant>
      <scheduledState>STOPPED</scheduledState>
      <schedulingStrategy>TIMER_DRIVEN</schedulingStrategy>
      <runDurationNanos>0</runDurationNanos>
    </processor>
    <processor>
      <id>3bbe0931-dc54-490e-96cf-23f503f8915a</id>
      <name>NoOperation</name>
      <position x="278.0" y="272.0"/>
      <styles/>
      <comment/>
      <class>com.etretatlogiciels.nifi.processor.NoOperation</class>
      <maxConcurrentTasks>1</maxConcurrentTasks>
      <schedulingPeriod>0 sec</schedulingPeriod>
      <penalizationPeriod>30 sec</penalizationPeriod>
      <yieldPeriod>1 sec</yieldPeriod>
      <bulletinLevel>WARN</bulletinLevel>
      <lossTolerant>false</lossTolerant>
      <scheduledState>STOPPED</scheduledState>
      <schedulingStrategy>TIMER_DRIVEN</schedulingStrategy>
      <runDurationNanos>0</runDurationNanos>
    </processor>
    <connection>
      <id>15af94b3-3db2-4bee-a7db-68869ad9bb86</id>
      <name/>
      <bendPoints>
        <bendPoint x="234.5" y="229.0"/>
      </bendPoints>
      <labelIndex>1</labelIndex>
      <zIndex>0</zIndex>
      <sourceId>3bbe0931-dc54-490e-96cf-23f503f8915a</sourceId>
      <sourceGroupId>9a3b3f31-d3fa-4e18-853f-f49510f48c2a</sourceGroupId>
      <sourceType>PROCESSOR</sourceType>
      <destinationId>c3daf696-11e7-4684-9e84-9f1aaedf15a6</destinationId>
      <destinationGroupId>9a3b3f31-d3fa-4e18-853f-f49510f48c2a</destinationGroupId>
      <destinationType>PROCESSOR</destinationType>
      <relationship>success</relationship>
      <maxWorkQueueSize>0</maxWorkQueueSize>
      <maxWorkQueueDataSize>0 MB</maxWorkQueueDataSize>
      <flowFileExpiration>0 sec</flowFileExpiration>
    </connection>
    <connection>
      <id>c180a307-c0f5-48b5-8816-9abc54962b5b</id>
      <name/>
      <bendPoints/>
      <labelIndex>1</labelIndex>
      <zIndex>0</zIndex>
      <sourceId>c3daf696-11e7-4684-9e84-9f1aaedf15a6</sourceId>
      <sourceGroupId>9a3b3f31-d3fa-4e18-853f-f49510f48c2a</sourceGroupId>
      <sourceType>PROCESSOR</sourceType>
      <destinationId>e1d7c27f-e22f-480d-9d8d-591accbb7180</destinationId>
      <destinationGroupId>9a3b3f31-d3fa-4e18-853f-f49510f48c2a</destinationGroupId>
      <destinationType>PROCESSOR</destinationType>
      <relationship>Stop retrying</relationship>
      <maxWorkQueueSize>0</maxWorkQueueSize>
      <maxWorkQueueDataSize>0 MB</maxWorkQueueDataSize>
      <flowFileExpiration>0 sec</flowFileExpiration>
    </connection>
    <connection>
      <id>ea3c8896-7775-4fce-83fe-8e95467c7811</id>
      <name/>
      <bendPoints/>
      <labelIndex>1</labelIndex>
      <zIndex>0</zIndex>
      <sourceId>0c689901-1e21-4a89-af21-04ec13f1ca03</sourceId>
      <sourceGroupId>9a3b3f31-d3fa-4e18-853f-f49510f48c2a</sourceGroupId>
      <sourceType>PROCESSOR</sourceType>
      <destinationId>c3daf696-11e7-4684-9e84-9f1aaedf15a6</destinationId>
      <destinationGroupId>9a3b3f31-d3fa-4e18-853f-f49510f48c2a</destinationGroupId>
      <destinationType>PROCESSOR</destinationType>
      <relationship>success</relationship>
      <maxWorkQueueSize>0</maxWorkQueueSize>
      <maxWorkQueueDataSize>0 MB</maxWorkQueueDataSize>
      <flowFileExpiration>0 sec</flowFileExpiration>
    </connection>
    <connection>
      <id>f5739172-bf92-4ef8-a41b-b6e4dd363468</id>
      <name/>
      <bendPoints/>
      <labelIndex>1</labelIndex>
      <zIndex>0</zIndex>
      <sourceId>c3daf696-11e7-4684-9e84-9f1aaedf15a6</sourceId>
      <sourceGroupId>9a3b3f31-d3fa-4e18-853f-f49510f48c2a</sourceGroupId>
      <sourceType>PROCESSOR</sourceType>
      <destinationId>3bbe0931-dc54-490e-96cf-23f503f8915a</destinationId>
      <destinationGroupId>9a3b3f31-d3fa-4e18-853f-f49510f48c2a</destinationGroupId>
      <destinationType>PROCESSOR</destinationType>
      <relationship>Retry operation</relationship>
      <maxWorkQueueSize>0</maxWorkQueueSize>
      <maxWorkQueueDataSize>0 MB</maxWorkQueueDataSize>
      <flowFileExpiration>0 sec</flowFileExpiration>
    </connection>
  </rootGroup>
  <controllerServices/>
  <reportingTasks/>
</flowController>

This file stores the flow which represents the processors on the canvas and the relationships between them. The record of processors (lines 10, 31, 62, 79 and 96) includes their properties, even sensitive ones (like passwords—though none here), as shown on lines 53-60 above, and their graphical position on the canvas (see lines 13, 34, etc. above).

Also highlighted above are the names of relationships that correspond to connections. Here's a second flow, this time with a process group:

Inside "MyProcessGroup"...

Unzipped (from flow.xml.gz), this flow looks like this:

<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<flowController encoding-version="1.0">
  <maxTimerDrivenThreadCount>10</maxTimerDrivenThreadCount>
  <maxEventDrivenThreadCount>5</maxEventDrivenThreadCount>
  <rootGroup>
    <id>746c58aa-0158-1000-08a8-9984ee3a72bc</id>
    <name>NiFi Flow</name>
    <position x="0.0" y="0.0"/>
    <comment/>
    <processor>
      <id>74a80bc0-0158-1000-8699-37fc9ff670e4</id>
      <name>CreateFlowFile</name>
      <position x="252.0" y="4.0"/>
      <styles/>
      <comment/>
      <class>com.etretatlogiciels.nifi.processor.CreateFlowFile</class>
      <maxConcurrentTasks>1</maxConcurrentTasks>
      <schedulingPeriod>0 sec</schedulingPeriod>
      <penalizationPeriod>30 sec</penalizationPeriod>
      <yieldPeriod>1 sec</yieldPeriod>
      <bulletinLevel>WARN</bulletinLevel>
      <lossTolerant>false</lossTolerant>
      <scheduledState>STOPPED</scheduledState>
      <schedulingStrategy>TIMER_DRIVEN</schedulingStrategy>
      <runDurationNanos>0</runDurationNanos>
      <property>
        <name>Content</name>
        <value>All good me must come to the aid of their country.</value>
      </property>
    </processor>
    <processor>
      <id>74a8e138-0158-1000-92f3-c6ec8781ba64</id>
      <name>NoOperation</name>
      <position x="249.0" y="601.0"/>
      <styles/>
      <comment/>
      <class>com.etretatlogiciels.nifi.processor.NoOperation</class>
      <maxConcurrentTasks>1</maxConcurrentTasks>
      <schedulingPeriod>0 sec</schedulingPeriod>
      <penalizationPeriod>30 sec</penalizationPeriod>
      <yieldPeriod>1 sec</yieldPeriod>
      <bulletinLevel>WARN</bulletinLevel>
      <lossTolerant>false</lossTolerant>
      <scheduledState>STOPPED</scheduledState>
      <schedulingStrategy>TIMER_DRIVEN</schedulingStrategy>
      <runDurationNanos>0</runDurationNanos>
    </processor>
    <processGroup>
      <id>74a6c4bd-0158-1000-3de4-adbe359fb278</id>
      <name>MyProcessGroup</name>
      <position x="232.0" y="289.0"/>
      <comment/>
      <processor>
        <id>74a70fd2-0158-1000-3d8e-645008570026</id>
        <name>NoOperation</name>
        <position x="285.0" y="184.0"/>
        <styles/>
        <comment/>
        <class>com.etretatlogiciels.nifi.processor.NoOperation</class>
        <maxConcurrentTasks>1</maxConcurrentTasks>
        <schedulingPeriod>0 sec</schedulingPeriod>
        <penalizationPeriod>30 sec</penalizationPeriod>
        <yieldPeriod>1 sec</yieldPeriod>
        <bulletinLevel>WARN</bulletinLevel>
        <lossTolerant>false</lossTolerant>
        <scheduledState>STOPPED</scheduledState>
        <schedulingStrategy>TIMER_DRIVEN</schedulingStrategy>
        <runDurationNanos>0</runDurationNanos>
      </processor>
      <inputPort>
        <id>74a74ed5-0158-1000-76fc-eeb723a5f47a</id>
        <name>to MyProcessGroup</name>
        <position x="228.0" y="74.0"/>
        <comments/>
        <scheduledState>STOPPED</scheduledState>
      </inputPort>
      <outputPort>
        <id>74a772d8-0158-1000-4916-01c5a8b496e9</id>
        <name>from MyProcessGroup</name>
        <position x="432.0" y="402.0"/>
        <comments/>
        <scheduledState>STOPPED</scheduledState>
      </outputPort>
      <connection>
        <id>74a79d27-0158-1000-ecf5-2c3ae0360366</id>
        <name/>
        <bendPoints/>
        <labelIndex>1</labelIndex>
        <zIndex>0</zIndex>
        <sourceId>74a70fd2-0158-1000-3d8e-645008570026</sourceId>
        <sourceGroupId>74a6c4bd-0158-1000-3de4-adbe359fb278</sourceGroupId>
        <sourceType>PROCESSOR</sourceType>
        <destinationId>74a772d8-0158-1000-4916-01c5a8b496e9</destinationId>
        <destinationGroupId>74a6c4bd-0158-1000-3de4-adbe359fb278</destinationGroupId>
        <destinationType>OUTPUT_PORT</destinationType>
        <relationship>success</relationship>
        <maxWorkQueueSize>10000</maxWorkQueueSize>
        <maxWorkQueueDataSize>1 GB</maxWorkQueueDataSize>
        <flowFileExpiration>0 sec</flowFileExpiration>
      </connection>
      <connection>
        <id>74a78f0c-0158-1000-a1f3-a3c2c8730ff2</id>
        <name/>
        <bendPoints/>
        <labelIndex>1</labelIndex>
        <zIndex>0</zIndex>
        <sourceId>74a74ed5-0158-1000-76fc-eeb723a5f47a</sourceId>
        <sourceGroupId>74a6c4bd-0158-1000-3de4-adbe359fb278</sourceGroupId>
        <sourceType>INPUT_PORT</sourceType>
        <destinationId>74a70fd2-0158-1000-3d8e-645008570026</destinationId>
        <destinationGroupId>74a6c4bd-0158-1000-3de4-adbe359fb278</destinationGroupId>
        <destinationType>PROCESSOR</destinationType>
        <relationship/>
        <maxWorkQueueSize>10000</maxWorkQueueSize>
        <maxWorkQueueDataSize>1 GB</maxWorkQueueDataSize>
        <flowFileExpiration>0 sec</flowFileExpiration>
      </connection>
    </processGroup>
    <connection>
      <id>74b5493a-0158-1000-617d-40ba69f7ad05</id>
      <name/>
      <bendPoints/>
      <labelIndex>1</labelIndex>
      <zIndex>0</zIndex>
      <sourceId>74a772d8-0158-1000-4916-01c5a8b496e9</sourceId>
      <sourceGroupId>74a6c4bd-0158-1000-3de4-adbe359fb278</sourceGroupId>
      <sourceType>OUTPUT_PORT</sourceType>
      <destinationId>74a8e138-0158-1000-92f3-c6ec8781ba64</destinationId>
      <destinationGroupId>746c58aa-0158-1000-08a8-9984ee3a72bc</destinationGroupId>
      <destinationType>PROCESSOR</destinationType>
      <relationship/>
      <maxWorkQueueSize>10000</maxWorkQueueSize>
      <maxWorkQueueDataSize>1 GB</maxWorkQueueDataSize>
      <flowFileExpiration>0 sec</flowFileExpiration>
    </connection>
    <connection>
      <id>74a8a54b-0158-1000-7902-ac61947fd255</id>
      <name/>
      <bendPoints/>
      <labelIndex>1</labelIndex>
      <zIndex>0</zIndex>
      <sourceId>74a80bc0-0158-1000-8699-37fc9ff670e4</sourceId>
      <sourceGroupId>746c58aa-0158-1000-08a8-9984ee3a72bc</sourceGroupId>
      <sourceType>PROCESSOR</sourceType>
      <destinationId>74a74ed5-0158-1000-76fc-eeb723a5f47a</destinationId>
      <destinationGroupId>74a6c4bd-0158-1000-3de4-adbe359fb278</destinationGroupId>
      <destinationType>INPUT_PORT</destinationType>
      <relationship>success</relationship>
      <maxWorkQueueSize>10000</maxWorkQueueSize>
      <maxWorkQueueDataSize>1 GB</maxWorkQueueDataSize>
      <flowFileExpiration>0 sec</flowFileExpiration>
    </connection>
  </rootGroup>
  <controllerServices/>
  <reportingTasks/>
</flowController>

Templates and process groups

The two artifacts that are crucial to our ability to deploy are

1. process groups
2. templates

Process groups are found inside flow.xml.gz, as shown above.

Tuesday, 22 November 2016

On a cluster node going down...

Wednesday, 30 November 2016

Imagine a conf/flow.xml.gz that runs as soon as NiFi is started. However, imagine it hanging and the UI is unavailable to stop the processors, look around or otherwise look for the problem. What to do?

In conf/nifi.properties, there is nifi.flowcontroller.autoResumeState that can be set to false (the default is true) before restarting NiFi. Look for errors in logs/nifi-app.log.

Remember to set back to the default when finished. If any changes in debugging the problem are made to the flow, a new conf/flow.xml.gz will be generated with processors in a stopped state (or with other changes made when debugging) which may not be what is wanted.

Thursday, 1 December 2016

To log what flowfile is getting passed out from each processor to understand what the processor has done to that flowfile, LogAttribute processor has a property, Log Payload, that can be set to true that you could use this at different points in your flow.

Another solution that will work without modifying a flow is to use NiFi's provenance feature from the UI. Right-click on a processor and select Data provenance, then pick an event and view the lineage graph. (Note that provenance in NiFi is typically configured to evaporate in order to save space, so there may be nothing to look at if you wait too long). From there you can look at the content of each event in the graph to see how it changed as it moved through the flow.

Rather than examining the lineage, you could clikc View Details (the "info" or question-mark icon to the left of the table record). Then switch to the Content tab to see the in-coming and out-going content at each step of provenance. Clicking Download will bring the raw output to your filesystem while View will open a content viewer in another window displaying the flowfile in various text formats.

UI notifications; processors faulting

One way to draw a user's attention is to use a standard log statement with proper level. This will issue a bulletin both on the processor (and visually indicate a note) as well as add it to a global list of bulletins. By default processors ignore anything lower than WARN, but this is adjustable in processor settings (right-click the processor in the UI, choose Configure, click the Settings tab and see Bulletin Level).

Friday, 9 December 2016

I saw a thread in the developers' forum discussing a counter and I wondered about using ProcessSession's counter in place of writing a special processor for Mat and Steve that counts documents:

void adjustCounter( String name, long delta, boolean immediate ) Adjusts counter data for the given counter name and takes care of registering the counter if not already present, i.e.:

session.adjustCounter( "Counter", 1, false );
session.transfer( flowFile, SUCCESS );

I need to figure out where to see this counter. I discovered it in the NiFi Counters summary which is in a different place in the UI depending on the NiFi version (showing both at right). It lists the particulars including the individual passage of flowfiles through a processor that adjusts the count and also totals for all flowfiles through a processor that calls adjustCounter(). It's confusing if the processor is used over and over again in a flow because it shows up in the list.

Here's a listing in which I explicitly made this call, as coded just above, in Timer and ran through the flow illustrated above (yesterday).

(The long ids in the list above are processor instance ids.)

However, if I remove this explicit call, nothing shows up in this list after running a flowfile completely through the four instances of Timer. So, some processor must explicitly make this call.

So, to interpret what's in the list above, note that:

1. The total number of processors called that bump "Timer" is 4.
2. Each of
   • Timer 1 (start)
   • Timer 1 (stop)
   • Timer 2 (start)
   • Timer 2 (stop)
   (which are all just different instances of Timer) is called one time.

To name the counter, a property could be used or the name of the counter could come from the processor's special name (as set in the Configure → Settings → Name field.

It remains to be seen which processors of our we wish to put a counter into and how we want to know to make the call.

• All processors? It would grossly pollute the counter list.
• Carefully select list of processors? It would impede surrepticious and emergency use.
• Only make call if processor "renamed"?
• Only make call if NAMED_COUNTER property set with name?
• Either of the above? What about NoOperation which is routinely renamed?

Tuesday, 13 December 2016

NiFi site-to-site experiment

Wednesday, 15 December 2016

I was trying to subject NiFi to the Java Flight Recorder (JFR). However, adding the needed arguments to conf/bootstrap.conf proved difficult. NiFi wasn't having them (as reported naively in the developers' forum).

There is no problem in NiFi per se. It was not really necessary to dig down into the code really because there's no problem there. It's just how java.arg.n is consumed that creates trouble. One must play around with the value of n when argument ordering is crucial. In the present case, I was able to turn on JFR using the following (in bold) in conf/bootstrap.conf:

#Set headless mode by default
java.arg.14=-Djava.awt.headless=true

# Light up the Java Flight Recorder...
java.arg.32=-XX:+UnlockCommercialFeatures
java.arg.31=-XX:+FlightRecorder
java.arg.42=-XX:StartFlightRecording=duration=120m,filename=recording.jfr

# Master key in hexadecimal format for encrypted sensitive configuration values
nifi.bootstrap.sensitive.key=

The point was that -XX:+UnlockCommercialFeatures must absolutely precede the other two options. Because of these arguments passing through HashMap, some values of n will fail to sort an argument before others and the addition of other arguments might also come along and disturb this order. What I did here was by trial and error. I guess NiFi code could man-handle n in every case such that the order be respected. I'm guessing that my case is the only or one of the very rare times options are order-dependent. I wouldn't personally up-vote a JIRA to fix this.

My trial and error went pretty fast using this command line between reassigning instances of n for these arguments. It took me maybe 6 tries. (Script bounce-nifi.sh does little more than shut down, then start up NiFi, something I do a hundred times per day across at least two versions.)

~/dev/nifi/nifi-1.1.0/logs $ rm *.log ; bounce-nifi.sh 1.1.0 ; tail -f nifi-bootstrap.log

Thursday, 22 December 2016

http://stackoverflow.com/questions/151238/has-anyone-ever-got-a-remote-jmx-jconsole-to-work

I'm trying to JMC to run the Flight Recorder (JFR) to profile NiFi on a remote server that doesn't offer a graphical environment on which to run JMC.

Here is what I'm supplying to the JVM (conf/bootstrap.conf) when I launch NiFi:

java.arg.90=-Dcom.sun.management.jmxremote=true
java.arg.91=-Dcom.sun.management.jmxremote.port=9098
java.arg.92=-Dcom.sun.management.jmxremote.rmi.port=9098
java.arg.93=-Dcom.sun.management.jmxremote.authenticate=false
java.arg.94=-Dcom.sun.management.jmxremote.ssl=false
java.arg.95=-Dcom.sun.management.jmxremote.local.only=false
java.arg.96=-Djava.rmi.server.hostname=10.10.10.92  (the IP address of my server running NiFi)

I did put this in /etc/hosts, though I doubt it's needed:

10.10.10.92   localhost

Then, upon launching JMC, I create a remote connection with these properties:

Host: 10.10.10.92
Port: 9098
User: (nothing)
Password: (ibid)

Incidentally, if I click the Custom JMX service URL, I see:

service:jmx:rmi:///jndi/rmi://10.10.10.92:9098/jmxrmi

At some point, weeks later, this stopped working. I scratched my head. It turns out that I had run an Ansible script that activated iptables and I had to open port 9098 through the firewall created:

# iptables -A INPUT -p tcp --dport 9098 -j ACCEPT

Tuesday, 3 January 2017

Watching NiFi come up. In conf/nifi.properties, nifi.web.http.port=9091...

~/dev/nifi $ ps -ef | grep [n]ifi
russ     27483     1  0 14:37 pts/4    00:00:00 /bin/sh ./nifi.sh start
russ     27485 27483  2 14:37 pts/4    00:00:00 /home/russ/dev/jdk1.8.0_112/bin/java \
          -cp /home/russ/dev/nifi/nifi-1.1.1/conf:/home/russ/dev/nifi/nifi-1.1.1/lib/bootstrap/* \
          -Xms12m -Xmx24m \
          -Dorg.apache.nifi.bootstrap.config.log.dir=/home/russ/dev/nifi/nifi-1.1.1/logs /
          -Dorg.apache.nifi.bootstrap.config.pid.dir=/home/russ/dev/nifi/nifi-1.1.1/run /
          -Dorg.apache.nifi.bootstrap.config.file=/home/russ/dev/nifi/nifi-1.1.1/conf/bootstrap.conf /
          org.apache.nifi.bootstrap.RunNiFi start
russ     27508 27485 99 14:37 pts/4    00:00:21 /home/russ/dev/jdk1.8.0_112/bin/java /
          -classpath /home/russ/dev/nifi/nifi-1.1.1/./conf
                    :/home/russ/dev/nifi/nifi-1.1.1/./lib/slf4j-api-1.7.12.jar
                    :/home/russ/dev/nifi/nifi-1.1.1/./lib/nifi-documentation-1.1.1.jar
                    :/home/russ/dev/nifi/nifi-1.1.1/./lib/nifi-nar-utils-1.1.1.jar
                    :/home/russ/dev/nifi/nifi-1.1.1/./lib/log4j-over-slf4j-1.7.12.jar
                    :/home/russ/dev/nifi/nifi-1.1.1/./lib/logback-core-1.1.3.jar
                    :/home/russ/dev/nifi/nifi-1.1.1/./lib/jul-to-slf4j-1.7.12.jar
                    :/home/russ/dev/nifi/nifi-1.1.1/./lib/jcl-over-slf4j-1.7.12.jar
                    :/home/russ/dev/nifi/nifi-1.1.1/./lib/nifi-framework-api-1.1.1.jar
                    :/home/russ/dev/nifi/nifi-1.1.1/./lib/nifi-properties-1.1.1.jar
                    :/home/russ/dev/nifi/nifi-1.1.1/./lib/logback-classic-1.1.3.jar
                    :/home/russ/dev/nifi/nifi-1.1.1/./lib/nifi-api-1.1.1.jar
                    :/home/russ/dev/nifi/nifi-1.1.1/./lib/nifi-runtime-1.1.1.jar
          -Dorg.apache.jasper.compiler.disablejsr199=true /
          -Xmx512m /
          -Xms512m /
          -Dsun.net.http.allowRestrictedHeaders=true /
          -Djava.net.preferIPv4Stack=true /
          -Djava.awt.headless=true /
          -XX:+UseG1GC /
          -Djava.protocol.handler.pkgs=sun.net.www.protocol /
          -Dnifi.properties.file.path=/home/russ/dev/nifi/nifi-1.1.1/./conf/nifi.properties /
          -Dnifi.bootstrap.listen.port=38751 /
          -Dapp=NiFi /
          -Dorg.apache.nifi.bootstrap.config.log.dir=/home/russ/dev/nifi/nifi-1.1.1/logs /
          org.apache.nifi.NiFi
~ # netstat -lptu
Active Internet connections (only servers)
Proto Recv-Q Send-Q Local Address           Foreign Address         State       PID/Program name
tcp        0      0 localhost:63342         *:*                     LISTEN      23923/java
tcp        0      0 localhost:47060         *:*                     LISTEN      1876/GoogleTalkPlug
tcp        0      0 gondolin:domain         *:*                     LISTEN      1238/dnsmasq
tcp        0      0 *:ssh                   *:*                     LISTEN      1124/sshd
tcp        0      0 *:35353                 *:*                     LISTEN      23923/java
tcp        0      0 localhost:6942          *:*                     LISTEN      23923/java
tcp        0      0 *:9091                  *:*                     LISTEN      27508/java
tcp        0      0 localhost:45608         *:*                     LISTEN      27508/java
tcp        0      0 localhost:58634         *:*                     LISTEN      1876/GoogleTalkPlug
tcp6       0      0 [::]:64628              [::]:*                  LISTEN      24068/java
tcp6       0      0 [::]:ssh                [::]:*                  LISTEN      1124/sshd
tcp6       0      0 [::]:44155              [::]:*                  LISTEN      24068/java
tcp6       0      0 localhost:38751         [::]:*                  LISTEN      27485/java
udp        0      0 *:mdns                  *:*                                 919/avahi-daemon: r
udp        0      0 gondolin:domain         *:*                                 1238/dnsmasq
udp        0      0 gondolin:ntp            *:*                                 28138/ntpd
udp        0      0 localhost:ntp           *:*                                 28138/ntpd
udp        0      0 *:ntp                   *:*                                 28138/ntpd
udp        0      0 *:49427                 *:*                                 919/avahi-daemon: r
udp        0      0 *:ipp                   *:*                                 17819/cups-browsed
udp6       0      0 [::]:mdns               [::]:*                              919/avahi-daemon: r
udp6       0      0 [::]:56356              [::]:*                              919/avahi-daemon: r
udp6       0      0 fe80::cc63:b04:481e:ntp [::]:*                              28138/ntpd
udp6       0      0 ip6-localhost:ntp       [::]:*                              28138/ntpd
udp6       0      0 [::]:ntp                [::]:*                              28138/ntpd
~ # lsof -i | grep 9091
firefox    3842   russ   79u  IPv4 103681669      0t0  TCP localhost:57558->localhost:9091 (ESTABLISHED)
firefox    3842   russ   81u  IPv4 103686787      0t0  TCP localhost:57562->localhost:9091 (ESTABLISHED)
firefox    3842   russ   86u  IPv4 103686788      0t0  TCP localhost:57564->localhost:9091 (ESTABLISHED)
firefox    3842   russ   90u  IPv4 103686789      0t0  TCP localhost:57566->localhost:9091 (ESTABLISHED)
java      27508   russ 1699u  IPv4 103667313      0t0  TCP *:9091 (LISTEN)
java      27508   russ 1730u  IPv4 103687885      0t0  TCP localhost:9091->localhost:57562 (ESTABLISHED)
java      27508   russ 1731u  IPv4 103687886      0t0  TCP localhost:9091->localhost:57564 (ESTABLISHED)
java      27508   russ 1732u  IPv4 103680156      0t0  TCP localhost:9091->localhost:57558 (ESTABLISHED)
java      27508   russ 1734u  IPv4 103687887      0t0  TCP localhost:9091->localhost:57566 (ESTABLISHED)

Friday, 6 January 2017

Tuesday, 10 January 2017

Working on a huge flow, examining performance with JMC/JFR, I have literally hundreds of millions of files that stack up at different queue points, but especially at the end. Ridding myself of all theses files from the queues take something like forever. I posted to ask if anyone had a quick way, but there is none. I tried a) creating a no-op processor drain, but that doesn't go much if at all faster than b) right-clicking the queue and choosing Empty queue.

I even tried a brute-force approach using a script I wrote sitting in my NiFi root:

#!/bin/sh
read -p "Smoke NiFi repositories (y/n)? " yesno
case "$yesno" in
  y|Y)
    ( cd flowfile_repository   ; rm -rf * )
    ( cd provenance_repository ; rm -rf * )
    ( cd content_repository    ; rm -rf * )
    ;;
  *)
    ;;
esac

...because removing everything from the content repository takes forever—hours for a couple of hundred million flowfiles.

This was predictable, but I wasn't even thinking about it before the accumulation became a problem. After letting the files delete over night, I came in a added a counting processor so I could just let them drain out as they arrived (instead of accumulating them in a queue for counting).

Note too that if the total number of flowfiles in any one connection exceeds the value nifi.queue.swap.threshold in nifi.properties, usually 20,000, swapping will occur and performance can be affected. So, by allowing files to accumulate, I was shooting myself in the foot because I was altering the very performance I was trying to monitor/get a feel for.

Wednesday, 11 January 2017

I'd never done this before.

Thursday, 12 January 2017

Some notes on tuning NiFi for high-performance.

nifi.properties
1. nifi.bored.yield.duration=10 millis —prevents processors that are using the timer-driven scheduling technology from using excessive CPU despite there being no work to do.
2. nifi.ui.autorefresh.interval=30 sec —can be used to make the browser UI refresh more often, but increases burden on the network. It doesn't affect the performance of NiFi itself.
3. There are two databases used by NiFi (under database_repository). One is a user database keeping track of user logins (in the case where running secured NiFi). The other is the history database that tracks all changes made to the graph (or workflow). Neither of these is very big (I'm seeing just over 2Mb). It's useful to consider relocating this repository elsewhere (that is, outside /opt/nifi), to simplify upgrading by allowing to retain the user and component history after upgrading.
4. The most likely problem in corrupting repositories is running out of disk space. For this reason, relocating the flowfile-, content- and provenance repositories outside of /opt/nifi is not a bad idea. The location is specified in nifi.properties via nifi.flowfile.repository.directory, etc.
5. As noted a couple of days ago, it's the content repository's disk that experiences the highest I/O impact on systems with high data volumes. NiFi queues stacking up can be a problem. nifi.content.repository.directory.name=path can be use, in fact, a whole list of them with different names, in nifi.properties to spread out this potentially mammoth repository.
6. In NiFi clusters, it's a good idea to use different names (just as above) for the repositories of the different NiFi nodes. Obviously, installing these nodes will not lead automatically to different names and keeping the same, default names works fine, but monitoring disk utilization within the NiFi UI will be easier if the names are different when lumped together in the system diagnostics window.
bootstrap.conf
7. The JVM memory settings control NiFi's heap should be greatly expanded from the shipping defaults (for example, from -Xms256m to -Xms8g).

Friday, 13 January 2017

I still haven't had time to set up a NiFi cluster which I want to do experimentally (before I ever have to do it because forced).

I came across a random, passing observation that Zookeeper, which is integral to erecting a NiFi cluster, generally recommends an odd number of nodes. I'm sure that, just as for MongoDB, this is about elections.

Tuesday, 24 January 2017

The perils of moving a processor between packages.

When you change the class (or package) name of a processor, you have effectively removed the processor from your build and added a new one. At that point, NiFi will create a "ghost" processor (in conf/flow.xml.gz) to take its place because the processor class no longer exists.

However, the properties are still retained, in case the processor is added back (reappears). But since NiFi cannot find it, it doesn't know whether the properties are sensitive or not. As a result, it marks all properties as sensitive because it was determined that it is better to share too little information than to share too much.

It's possible to edit conf/flow.xml.gz by hand, if care is taken, I think.

Thursday, 2 February 2017

If disk space is exhausted, how to recover unprocessed data since NiFi cannot proceed?

Archived content can be deleted from content_repository:

# find content_repository/ -type f -amin +600 | grep archive | xargs rm -f

(The -amin +600 option means file was accessed 600 or more minutes ago.)

This deletes anything from that archive that is more than 10 hours old. Once this has happened, assuming enough diskspace has been freed up, NiFi can be restarted.

Of course, what partition both the content- and flowfile repositories are on is a consideration. If not the same one, then freeing up old content won't do the trick.

Note that it's theoretically safe to perform these deletions with NiFi running, but there are no guarantees. It's better to halt NiFi first, then restart it.

Saturday, 11 February 2017

Time out for NiFi performance...

I need to review some NiFi advice to see where we are on implementing NiFi the correct way. Here are some elements:

Saturday, 14 February 2017

When you get the error that a flowfile is not the most recent version, it means that you have modified the flowfile in some way via ProcessSession and then attempted to use the flowfile in a variable that preceeds the change. Commonly this happens when you have code that looks something like this:

FlowFile flowfile = session.get();

if( flowfile == null )
  return;

session.putAttribute( flowfile, "hello", "hello" );
session.transfer( flowfile, SUCCESS );

The issue here is that the ProcessSession.putAttribute() method called above returns a new FlowFile object, and we then pass the old version of the flowfile to session.transfer(). This code would need to be updated to look like this:

FlowFile flowfile = session.get();

if( flowfile == null )
  return;

flowfile = session.putAttribute( flowfile, "hello", "hello" );
session.transfer( flowfile, SUCCESS );

Note the flowfile variable is re-assigned to the new FlowFile object when session.putAttribute() is called so that we pass the most up-to-date version of the flowfile to ProcessSession.transfer.

Friday, 17 February 2017

On sensitive-value properties...

This is a phenomenon common in controller services, i.e.: properties that hold a password or -phrase or other sensitive sequence that must be hidden from subsequent view. You type it in, confirm it, and never see it again. You can only type it in again.

Here's a long comment I got from Andy LoPresto:

Each sensitive property value is encrypted using the specified algorithm noted in conf/nifi.properties. By default, and therefore on any unmodified system, this is PBEWITHMD5AND256BITAES-CBC-OPENSSL, which means that the actual cipher applied is AES with CBC mode of operation using a 256-bit key derived from the provided nifi.sensitive.props.key value (really a password and not a true key) using the OpenSSL EVP_BytesToKey key-derivation function (KDF), a modified version of PKCS #5 v1.5' MD5-based PBKDF1 identical when the combined length of the key and IV is less than or equal to an MD5 digest and non-standard if greater.

A random 16-byte salt is generated during the encrypt process (the salt length is actually set to be the block length of the cipher in question—AES is always 16 bytes, but if you choose to use DES for example, this value would be 8 bytes), and this salt is provided, along with an iteration count (1000 by default) to the KDF. Internally to the application, this process occurs within the Jasypt library (slated for removal in NIFI-3116), but you can see a compatible Java native implementation of this process in the ConfigEncryptionTool.

If the nifi.sensitive.props.key value is empty, a hard-coded password is used. This is why all documentation should recommend setting that value to a unique and strong passphrase prior to deployment.

Once the key is derived, the sensitive value is encrypted and the salt and cipher text are concatenated and then encoded in hexadecimal. This output is wrapped in "enc{" and "}" tokens to denote the value is encrypted, and then stored in flow.xml.gz. As pointed out elsewhere, the sensitive value (in plain-text or encrypted form) is never sent over the ReST API to any client, including the UI. When editing a processor property that is sensitive, the UI displays a static placeholder.

If two values are the same (i.e. the same password in an EncryptContent processor encrypting and an EncryptContent processor decrypting later in the same flow), the two encrypted values will be completely different in the flow.xml.gz even though they were encrypted with the same key. This is because of the random salt value mentioned above.

The encrypt-config.sh script in nifi-toolkit exposes the functionality to "migrate" the flow.xml.gz encrypted values to be encrypted with a new key. For example, if you read the above and feel uncomfortable with your values encrypted with the default hard-coded password rather than a unique passphrase, you can run this tool and provide a new passphrase, and it will update conf/nifi.properties with the new passphrase (and encrypt it if you have encrypted configuration enabled) and decrypt all values in the flow.xml.gz and re-encrypt them with the new key and repopulate them.

A small note on the migration tool—you may notice that after running it, identical values will have identical encrypted values. This design decision was made because the performance tradeoff of not re-performing the KDF with a unique salt and cipher initialization on each value is immense, and the threat model is weak because the key is already present on the same system. This decision was further impacted by the process by which Jasypt derives the keys. After NIFI-3116 is completed, I feel confident we can improve the security of the cipher texts by using unique IVs without incurring the substantial penalty currently levied by the library structure.

Throughout this explanation, you may have had concerns about some of the decisions made (algorithms selected, default values, etc.). There are ongoing efforts to improve these points, as security is an evolving process and attacks and computational processing availability to attackers is always increasing. More of this is available in relevant JIRAs and the Security Features Roadmap on the wiki, but in general I am looking to harden the system by restricting the algorithms used by default and available in general and increasing the cost of all key derivations (both in time and memory hardness via PBKDF2, bcrypt, and scrypt). Obviously the obstacles in this effort are backwards compatibility and legacy support. It is a balance, but with the support of the community, I see us continuing to move forward with regards to security while making the user experience even easier.

Wednesday, 1 March 2017

Tuesday, 7 March 2017

How the provenance repository works

Wednesday, 15 March 2017

If you want to see when/who stopped a processor in NiFi (and other configuration changes) you can use the Flow configuration history (the clock icon on the right side of the menu bar in 0.7.x) and filter by processor id.

Thursday, 16 March 2017

Just a note to remember that AbstractProcessor extends AbstractSessionFactoryProcessor and that the real delta between them is the second argument to onTrigger(), where the subclass offers ProcessSession in place of the ProcessSessionFactory. The latter gives the ability to create a new session (with sessionFactory.createSession().

Friday, 17 March 2017

Here are some excellent details on NiFi-processor semantics by Andy LoPresto and Mark Payne. First Andy...

Friday, 18 March 2017

I asked the question,

Mark Payne and Matt Burgess responded, Mark wrote:

Matt:

Tuesday, 21 March 2017

@SupportsBatching leads to that "slider" seen in a processor's configuration. It's a hint only, but has no material effect upon how to code the processor (in terms of special things to do).

A greater than 0 delay (see the slider) will cause the flowfiles over that time period to be batched out such that the work will be done, but no commit until the end of the delay (in slider).

Friday, 24 March 2017

Custom registry properties in the NiFi Expression Language

At least by NiFi 1.1.1, the following became a possibility:

In ${NIFI_ROOT}/conf/nifi.properties, add this line:

nifi.variable.registry.properties=./conf/registry.properties

You can list more than one file, separated by commas. Then, create this file (registry.properties) locally with contents:

zzyzzx="This is a test of the Emergency Broadcast System. This is only a test."

Next, create a flow using CreateFlowFile (and start it) connected to UpdateAttributes, which you configure thus:

Property  Value
ZZYZZX         ${zzyzzx}

...(starting this processor too) then on to NoOperation (that you don't start) so there's a flowfile in the queue you can look at. Bounce NiFi because these properties are static. Once there's a flowfile in the queue in front of NoOperation, right-click and choose List Queue, then click the View Details and, finally, the Attributes tab. You'll see attribute ZZYZZX with the property value assigned by registry.properties in our example.

Monday, 27 March 2017

Failing to restart NiFi...

Only the flowfile_repository need be removed (as long as data is not important): removing that repository means that the contents of the content_repository will be removed by NiFi since no flowfiles point to its contents anymore.

It's possible also/instead of to set nifi.flowcontroller.autoResumeState in conf/nifi.properties to false, then piecewise alleviate large queues or processing of files that could be causing the failure to restart.

Tuesday, 28 March 2017

You create a new processor, everything's perfect, then you move it out of your larger project to its own home ('cause that's needful), but then you cannot load it. The NAR builds, NiFi's happy to load it, but when you go to do Add Processor in the UI, it's not there. You pull your hair out, you cry on the shoulder of sympathetic souls in the community forum, then someone points out that you spelled META-INF with an underscore instead of a hyphen.

What's left to do, but look for a sword to fall on?

Here's a list of things to check that I will add to as more ideas occur to me.

1. Processor code builds and tests.
2. Processor contains a @Tags annotation; this won't stop it from being found in the UI list for Add Processor, but not having tags means having to run the list looking for it instead of being able to get nearly to it using a tag.
3. src/main/resources/META-INF/services/org.apache.nifi.processor.Processor accurately records the package path and processor name.
4. Check all package paths and names to ensure no conflicts with other processors.
5. Ensure a proper NAR is built:
   1. Your code is JAR'd up correctly: the JAR contains
      • the expected package path down to and including all the .class files of your project code.
      • a META-INF subdirectory containing the services subdirectory plus the file org.apache.nifi.processor.Processor with the definition of your processor and its accurate packagage name.
      • the file, MANIFEST.MF.
      • a maven subdirectory underneath which the pom.xml of your project code. (This is not the nar subdirectory's pom.xml nor the top-level one, but the one building the significant Java code.)
   2. Your JAR is NAR'd up: the NAR contains a META-INF subdirectory underneath which are:
      • your JAR, alongside all the myriad supporting JARs under the bundled-dependencies subdirectory.
      • the file, MANIFEST.MF.
      • a maven subdirectory underneath which the pom.xml for the nar subdirectory, that is, the instructions on how to build the NAR and what to include in it, its dependency(ies).
6. Copy the NAR to the lib subdirectory of NiFi.
7. Bounce NiFi; observe logs/nifi-app.log for trouble with the new processor.
8. Ensure you can find and add the new processor using the UI.

Monday, 3 April 2017

Bryan Bende makes some useful points in a post. I am going to be dealing with moving our authentication strategy from 0.7.1 over to 1.1.2 very soon, so I'm copying this post here for safe-keeping and quick access.

Wednesday, 5 April 2017

It looks like using NiFi 1.1.2 incurs a problem until 1.2.0 in that Jetty uses /dev/random. This generator will, the first time used on a machine, spend up to 10 minutes accumulating data from noise in order to be completely random. This would be the case every time in a launch of a virtual machine (since every launch is brand new "hardware"), so a JVM option can be used to change this. To override the use of /dev/random, set java.security.egd to use /dev/urandom. However, at least in the Oracle JVM, that string is noted and reinterpreted to be /dev/random thus thwarting the override. Instead use this incantation to get around it:

java.args.N=-Djava.security.egd=file:/dev/./urandom

...in ${NIFI_ROOT}/conf/bootstrap.conf.

Friday, 7 April 2017

Some notes on latency and through-put...

Some processors have, under the Scheduling tab of the Configure Processor dialog, a slider, Run Duration, that permits a balance to be set between latency and through-put.

This slider appears on processors making use of the @SupportsBatching annotation.

The way to think about it is to see the slider as representing how long a processor will be able to use a thread to work on flowfiles (from its in-bound queue), that is, allowing onTrigger() to run more times to generate more out-bound flowfiles. Moving the slider further to the right makes the processor do more work, but at the same time, the processor will hog the thread for a longer period of time before it gives it up to another processor. Overall latency could go down because flowfiles will sit in other queues for longer periods of time before being serviced by their processor (that is, the processor waiting to serve the queue they're sitting in) because other processors don't get the thread that the first one is hogging.

For processors that are willing to delegate the responsibility of when to commit what they've finished to the framework, in a transactional sense, the framework can then use that understanding to combine one or more into a single transaction. This trades off some small latency for what is arguably higher throughput because the framework can do a single write to the flowfile repository in place of doing many writes reducing thus the burden on various locks, filesystem interrupts, etc. In some cases, this is more friendly and has the effect of higher through-put.

If a flow has and handles flows from numerous, perhaps competing concerns, teams, organizations, etc., this must be tunable.

Whether or not to annotate with @SupportsBatching, enabling the existence of this slider, it comes down to whether the processor can or is willing to let the framework control when to commit the session's work. This would be the case when work isn't side-effect free (cf. @NoSideEffects) in which something, some state, has been altered in ways from which actions taken could not be recovered from. For example, PutSFTP will send data via SFTP. Once a file is sent, it cannot be "unsent" and the process cannot be repeated without side effect. This processor would not be a candidate for allowing the framework to decide.

Tuesday, 18 April 2017

I had the opportunity of witnessing what happens when a NAR is inadvertantly restructured resulting in processors ending up on different package paths to where they were before. Of course, they no longer work as they are no longer there in support of processors in existing flows. NiFi makes "ghosts" when that happens and, of course, those ghosts don't work. I was impressed, however, by what happened. I half-expected some cataclysm with damage to flow.xml.gz. The only thing that happened was an exception in logs/nifi-app.log, a comment about the ghost created, then in the flow, the processors weren't operable, but they were still there and still connected.

When I corrected the situation, everything worked again.

Wednesday, 19 April 2017

NiFi processor properties

Today I'm going to corral the topic of responding to changes in configuration. Typically, a processor's properties will remain fixed once established, but it doesn't have to be that way. I'm especially interested in optimizing gathering properties since some properties might be expensive to gather if they need further treatment after gathering.

It's desirable, therefore, for a processor to react when its properties have changed. The method, onPropertyModified(), is inherited from AbstractProcessor and is called when these have changed, once for each modified property:

/**
 * Called only when a NiFi user actually updates (changes) this property's value.
 * @param descriptor the property.
 * @param oldValue null if property had no previous value.
 * @param newValue null if property has been removed.
 */
@Override
public void onPropertyModified( final PropertyDescriptor descriptor, final String oldValue,
                                                    final String newValue )

NiFi component lifecycle (and annotations)

This brings up also NiFi's component lifecycle, which is governed by means of annotations. Methods are annotated to be called when the lifecycle event occurs.

• @OnAdded —called whenever a component (like a controller) is created. This is rather static therefore. This event is also when the init() method, from AbstractProcessor is called.
• @OnEnabled —called whether a controller is enabled.
• @OnRemoved —called whenever a processor is removed from the flow so any freeable resources may be cleaned up.
• @OnScheduled —called every time the processor (not for controllers, but for reporting tasks and processors) is run so any processor resources needing allocation for the one call to onTrigger() may be constructed. This is a good time to check for new dynamic properties.
• @OnUnscheduled —called every time the processor the processor (or reporting task) is no longer scheduled to be run for cleaning up what was constructed.

So, what does this mean for property management?

It means you define static properties (and relationships) in your init() method. However, dynamic properties are not there at the time init() is called since they're dynamic and you can't predict them.

If you expect multiple instances of your processor (duh), you'll want to use instance variables to hang on to your properties. You'll also want instance variables, in the case of expensive, derivative work fo the sort that prompted me to think about this today in the first place (because I've not previously had this challenge of expensive operations as a result of property values that could change), so that each instance of your processor has and deals with this expense separately. (Of course, you shouldn't use static variables for stuff like this.)

The most efficient way to react only to expensive decisions is to watch @OnScheduled. That's the time that your processor is about to run, you want to lock down all the meaningful data then rather than keep doing it over and over again each time onTrigger() is called.

If you've got any deconstruction concerns, which would be rare, you'd want to handle those in an @OnStopped setting. Don't do this in an @OnUnscheduled method because the processor threads aren't halted by then and you'd be pulling the rug from under their feet.

Last, if you want precision reaction only to those properties reported as modified, you can finesse it by writing a method I evoked at the beginning, onPropertyModified().

For the present work, some of this has proven useful and I'm content.

Important note in @OnSceduled methods: It's important here that unvalidated properties, in this case only the rules from dynamic properties, don't lead (when missing or bad) to a situation that result in exceptions, errors, etc. that aren't carefully handled (by clear warning). Otherwise, no sooner is the processor started up on the UI canvas than it gets a potentially confusing error/warning notation on it.

Thursday, 20 April 2017

Clustering notes...

...from comments by Mark Payne.

NiFi relies upon Apache Zookeeper to coordinate which node acts as coordinator and which as primary. An embedded Zookeeper is an option, but for production use, an external one running on different hosts is recommended.

For getting started, the nifi.cluster.node.connection.timeout and the nifi.cluster.node.read.timeout properties are set to 5 seconds by default, but garbage collection for large numbers of flowfiles could cause pauses longer than that, so it's recommended to increase these values to 10 or even 15 seconds. Similarly, change nifi.zookeeper.connect.timeout and nifi.zookeeper.session.timeout to 5 or 10 seconds.

And, something else important (if also a bit "duh?!") I discovered. Unless something is specifically done to make nodes in a NiFi cluster exchange data in a flow (and this isn't for the faint-hearted), a flowfile will always remain on the same node that created it, beginning to end.

Later note: prior to NiFi 1.8, it was necessary to use a remote-process group connected to the cluster via an input/output port plus the site-to-site protocol to load-balance data running through a NiFi cluster. Beginning with 1.8, however, it became possible to configure this in the relationship between any two processors.

Monday, 24 April 2017

Counter-intuitively, here's how to export, then import a template. I say this because I don't find this interface all that compelling and could come up with a more logical one myself. Nevertheless, we're certain happy that this is possible at all.

The first sequence is exporting the template while browsing one NiFi flow. The second sequence shows importing the template while browsing another (or the same) NiFi flow. Finally, the last sequence shows incorporating the contents of the template as a new NiFi flow.

Save elements of a flow to a new template...
1. Select the elements of your flow that are to be the template.
2. From the Operate menu, click the Create Template button/icon (above).
3. Name and Describe (if desired) the template.
4. The template is now in conf/flow.xml.gz (in NiFi 0.x it was kept in a separate file under conf/templates). To export it as .xml, pull down the General menu (upper right-hand corner of NiFi's content window in the browser) and choose Templates.

5. Click to select your template.
6. Click the Download icon (right), click the Save radio button, then OK, then navigate to where you wish to drop it and click Save.
7. Note: it's ironically easier for the next steps to keep this template on the host from which you run the browser hosting the NiFi UI.
Import the new template to the target flow...
8. From the Operate menu again, click the Upload Template button/icon (above).
9. Click the Browse (magnifier) button and and navigate to the template you saved above, then click Open. You'll see the template name in the Upload Template dialog.
10. Click the UPLOAD button. You'll see a confirmation alert. Dismiss it by clicking OK. At this point, you can look via General → Templates to see your new template listed. If you can't see it there, then you will not be able to import it in this next sequence of steps.
Place the template's contents on the canvas...
11. Wherever you want to insert this template, pull a template from the Components toolbar (below) and drop it on the canvas where you want it.
12. Before it drops, you'll have the opportunity to select your template using the Choose Template drop-down in the Add Template dialog. Select it and choose Add. If any processors are missing (i.e.: the instance of NiFi you're modifying isn't the same version or, especially, doesn't have the same NAR files (in lib), you won't be able to drop the new template.

Tuesday, 25 April 2017

Yesterday, I learned (or reinforced having learned it before) that our searchadmin user cannot touch the NiFi canvas/flows—only examine it (them). This is because, sometime in the history of NiFi 1.x, it became the case that a user is created and able to create other users then endow them with the ability to do this, but not do it (not by default, that is) by itself.

The symptom of this was, because I was logged in from a user with rights to assign rights, but without rights to edit the flow, much of the UI was greyed out.

So, one thing I had to do yesterday was to create a new user, russ, on the test server (above) that can create, modify and delete NiFi flows. Here are some notes on this using a new user, jack. A new user is created in LDAP whence it's consumed by NiFi. Then, manage NiFi this way:

1. In NiFi, log in as searchadmin. (This user of ours is not able to do more than view the NiFi canvas, but it's the user that can manage other users and groups too.)
2. In the General menu, select Users then:
   1. Find an existing user and copy his fully qualified distinguished name (FQDN) to your clipboard. This would be something like "uid=russ,ou=People,dc=searchappliance,dc=com". The first time, however, if there isn't an example to use, you'd have to find out how FQDNs are already set up in your LDAP for this.
   2. Click on the Add User button/icon at the extreme right.
   3. Paste (and correct to have the right user name) the FQDN you want to add, something like "uid=jack,ou=People,dc=searchappliance,dc=com" into the Identity field. Don't assign ROLE_ADMIN.
   4. Click OK.
(Note: I'm not dealing with the concept of a user belonging to groups here.)
3. Copy the new FQDN you created to the clipboard and dismiss the NiFi Users dialog.
4. Open the Operate tool palette and click the Access Policies dialog box via the key icon or button. (However, see also my note in the last step of this post.)
5. Add the new user to each access policy he'll need. This is pretty tedious. Here's how:
   1. Click the Add users/groups to this policy (the first policy is view the component.
   2. To add the new user in the Add Users/Groups dialog, click in the User Identity field.
   3. Paste what you previously copied to the clipboard, the FQDN of your new user, then select it (as it's reproduced just below that field) so that it shows up in the Selected Users list. (I found this step very unwieldy and confusing; you'll need to experiment with this odd UI control until you get it to show up in the list.)
   4. Click Add. What this should accomplish is that the new user is added to the list of users able to view the component.
   5. Click the drop=down where you see view the component and select a different policy, like modify the component.
   6. Click the Add users/groups to this policy button/icon.
   7. Click in the User Identity field, a list should appear; select your new user. (You may need to paste again from the clipboard—as I say, I find this very unwieldy.) The point is to add the new user to the Selected Users list.
   8. Do this for every policy such that the new user has rights to accomplish all that he's expected to do in the NiFi UI in terms of creating and managing flows. Here are the policies that I set up for jack:
      • view the user interface
      • access the controller: view, modify
      • access restricted components
      • retrieve site-tosite details
      • view system diagnostics
      • proxy user requests
      • access counters: view, modify
      What don't I have access to?
      • query provenance (no way to set this up?)
      • access all policies (would make jack a super user)
      • access users/user groups (ibid)
   9. If you're being careful, this new user you're adding probably doesn't need the ability to view- or modify the policies.
   10. Dismiss the Access Policies dialog.
6. Log out from searchadmin and log in using the new user. If you get:

   Access Denied
   
   Unable to perform the desired action due to insufficient permissions. Contact the system administrator.
   

   you've neglected something. You'll need to log out, log back in as searchadmin, and fix it. Likely, the user doesn't have adequate privileges to view the user interface. It's always better to manage the policies via General → Policies than via the Operate → Access Policies method. For some reason, you don't see all the policies via the second method which is the one I suggested in these steps. Sorry.

Note that, while this is overly complicated and nasty, there is a way to work around it. To do this, create a new group that can read and write the UI canvas, then just add users into that group. Note that it need not be an LDAP group, only a group for convenience in NiFi.

Thursday, 27 April 2017

The behavior of creating a property as a dynamic one (e.g.: creating using the builder with method .isDynamic( true ) ) does lead to a nice, dynamic-looking property in the configuration dialog replete with a trashcan icon/control, but deleting the property doesn't in fact delete it: the next time you open the configuration dialog, it's back and, anyway, it remains in the list of properties to consider when you iterate through them, i.e.:

for( Map.Entry< PropertyDescriptor, String > property : context.getProperties().entrySet() )
{
  PropertyDescriptor descriptor = property.getKey();

  if( descriptor.isDynamic() )
    ...
}

Separately, in JUnit testing a processor, you add dynamic properties in the same way as static ones for the purpose of testing, so this doesn't inform you much in the behavior noted in my first paragraph.

Joe Witt said that the dynamic-property mechanism wasn't really set up to work for what I want to do (preconfigure some very likely defaults to speed up what my users have to do). The answer is to pre-add properties with a boolean enabling property and let them turn them off instead. In my precise need, this works fine though I'm pretty sure the idea has some merit.

Friday, 28 April 2017

Backing up a NiFi instance...

My IT guy is interested in back-ups of NiFi, so I thought I'd do a little research on it. Here are some notes of my own and some taken from the research I'm doing this morning. I haven't been overly interested in this aspect of NiFi in the year-plus I've been writing processors to solve problems using this framework.

• Reproducing an instance of NiFi vs. Restoring the state of a running instance of NiFi
  These are important distinctions because the answers are different, the former is very easy to do, the latter more challenging.
  
  

  It's important to note that restoring the running state of an instance including being able to reproduce the instance since restarting NiFI with content (live data) that doesn't match the flow definition is icky at best. The most encouraging thing that can be said is that doing this puts NiFi in an undefined state and you can expect undefined results from it.

• conf/nifi.properties and conf/bootstrap.conf
  These files keep configuration data for NiFi, in particular for the question of backing it up, it's possible to modify the locations of nearly all the subdirectories and files that figure in this discussion. This fact is important when considering the last point in my list here.
  
  

  On a NiFi cluster, there are additional configuration files that would need to be backed up, in particular, those associated with Zookeeper, conf/zookeeper.properties.

• conf/flow.xml.gz
  The entire flow for a NiFi instance is stored in conf/flow.xml.gz. It's kept constantly up-to-date in that, as soon as anyone changes anything on the canvas, it's already in this file. This file is created, if missing, as soon as NiFi starts up.
  
  

  Since NiFi 1.0, templates have begun to be collected in this file. Previously, they were kept in conf/templates.

• bin
  Any custom scripts kept here would be above and beyond what ships in the NiFi download and might be candidates for back-up.

• lib
  Any custom processors in use would be above and beyond what ships in the NiFi download and might be candidates for back-up.

• work
  There is nothing to archive here. This is where everything in /lib is exploded prior to NiFi running.

• logs/*.log
  The logs for NiFi, in particular:
  • nifi-bootstrap.log, what happens at start-up and restart, also, the default location of Java garbage-collection activity if logging is turned on for that.
  • nifi-user.log, precious little of interest here, only statements about user logging in, attempts to access the instance from web browsers, etc.
  • nifi-app.log, the nitty-gritty of what's going on at the microsecond interval including INFOrmation statements, WARNings, ERRORs, DEBUGging, and TRACEing depending on what logging levels are set, etc.
  • el-audit.log. or, in other words, custom log files, as created by and for the instance.
  • Note: If you start NiFi with no log files, these will be created. If you remove any of these log files while NiFi is running, they are neither recreated nor written to until NiFi is restarted.

• database_repository
  There is little say here. This is where users and groups are kept. I'm not sure what else.

• flowfile_repository
  This is where the flowfiles themselves (not their contents) are recorded, mostly their metadata including attributes. The flowfile repositiory is implemented as a write-ahead log. It keeps appending updates to the files until it check-points (every 2 minutes by default). If there are a lot of flowfiles, those updates can take up a lot of disk space. Also, a lot of space if those flowfiles happen to carry a lot of attributes and attribute data. Use nifi.flowfile.repository.checkpoint.interval in conf/nifi.properties to make NiFi check-point this repository more frequently. This repository is rarely huge because of check-pointing.

• content_repository
  In short, the current contents of the flowfiles. This collection of subdirectories and files is by far the biggest chunk of diskspace proportionately. nifi.properties permits you to relocate and split this repository over multiple volumes (and it's very wise to do so).

• provenance_repository
  Provenance is a record of where flowfiles have been to assist in the forensics of answering the question, "What happened to my data?" This repository can be split up too.

• One NiFi instance I read about divided up diskspace this way:
  • content_repository: 8 volumes of 200Gb each
  • flowfile_repository: 1 volume of 50Gb (I was a little shocked by this small size, but I later learned from Mark Payne that 30Gb is quite large.)
  • database_repository: 1 volume of 25Gb
  • provenance_repository: 2 volumes of 200Gb each (meh...who cares?)
  ...and also:
  • /opt/configuration-resources: 25Gb (I'm assuming conf/*)
  • /var/log/nifi-logs: 100Gb
  • /opt/nifi: 50Gb (doesn't include the foregoing, obviously)

Mark Payne offered this advice:

Monday, 1 May 2017

Flowfile content

As long as the content of a flowfile doesn't change, but only attributes, only a reference is maintained to it—not separate copies.

Tuesday, 2 May 2017

To turn on logging to get help with a standard NiFi processor, let's pick InvokeHTTP, paste this line into (toward the bottom) ${NIFI_ROOT}/conf/logback.xml:

<logger name="org.apache.nifi.processors.standard.InvokeHTTP" level="DEBUG"/>

Monday, 8 May 2017

If your unit test needs to check for cross-contamination by simulating two, consecutive runs of your processor, you instantiate a new TestRunner only once, for sure, but after checking out the results from the first call to runner.run( 1 ), you need to call runner.clearTransferState() to put the "processor instance" back in the state it would be (fresh) for the next call.

Friday, 12 May 2017

More on slf4j trouble...

I found by accident and more specifically what's troubling me in my case.

Of course, the real problem is having older slf4j JARs linked into your binary because of ancient code. That's the real problem and the rest of this thread was taken up with getting to the real problem. However, ...

...now I know why not all of my test code encounters this problem. It's because it only happens when I use NiFi ComponentLog (or the test framework uses it). As long as execution never wanders down into that, I don't have the problem. It's the call to getLogger().{info|warn|etc.} that does it.

So, if you don't make any mistakes in your test code calling into the NiFi test framework, you're a lot less likely to encounter this. You also must not put a call to getLogger().{info|warn|etc.} into your custom-processor code.

Now, oddly, that NiFi call or my custom processor call getLogger().info(), etc. does not exhibit this problem in production, only when JUnit tests are running.

Makes an old C hack like me wish for a preprocessor. I wrote a class, NiFiLogger, to work through and that I disable in favor of System.out.println() when running tests. Down the road I'll be able to take that out.

Stuck threads

You have a processor that appears to be stopped, but also claims to be running two threads.

This is a stuck thread and there's no way, using the UI, to unstick the threads. You can bounce NiFi.

First, however, you can do

$ bin/nifi.sh dump

...to cause NiFi to do a dump to logs/nifi-bootstrap.log to find out why the threads are stuck.

Tuesday, 16 May 2017

NiFi state management

There's a simple, key-value pair store available for processors, controller services and reporting context, that allows for cluster-wide or local storage of state.

For example, if you want to remember what the last record you got doing some query, and you're the author of the processor doing it, you can add state to do that. The API consists of:

1. setState(),
2. replace(),
3. clear() and
4. getState().

(The implications of cluster state differ considerably from local state, so read up on this.)

Two processors cannot share state; it's per-processor only. To share state between processors, resort to a controller service for this, that is, to bind two processors to the same controller service and that service exposes thus its (otherwise) private state effectively shared between the two consuming processors.

Wednesday, 17 May 2017

I worked on the "NiFi rocks!" example of a controller service to excerpt (or "highlight") more or less the crucial bits because I wrote a controller service this week, had some trouble getting it to work just right (especially creating unit tests for it).

Here is the interface. Oddly, this is the very thing by which the controller service is always referenced (instead of StandardPropertiesFileService). In short, this interface must be able to say it all. Anything it doesn't say won't be easily accessible. Here, it says that the implementing controller service must make a method, String getProperty( String key ) available. Anything else the service does is immaterial: this method is the sole article in the contract. I have highlighted lines that are crucial to stuff in order to work.

PropertiesFileService.java:

package com.etretatlogiciels.nifi.controller.interfaces;

import org.apache.nifi.controller.ControllerService;

public interface PropertiesFileService extends ControllerService
{
  String getProperty( String key );
}

The controller service that extends PropertiesFileService just does that, implements what it does completely behind the contractual String getProperty( String key). Sure, onConfigured() does all the heavy lifting when called, but just sets up what String getProperty( String key ) will return when called by the consuming NiFi processor configured to use the controller service.

StandardPropertiesFileService.java:

package com.etretatlogiciels.nifi.controller;

import com.etretatlogiciels.nifi.controller.interfaces.PropertiesFileService;

@CapabilityDescription( "Provides a controller service to manage property files." )
public class StandardPropertiesFileService extends AbstractControllerService implements PropertiesFileService
{
  private String     configUri;
  private Properties properties = new Properties();

  @OnEnabled
  public void onConfigured( final ConfigurationContext context ) throws InitializationException
  {
    configUri = context.getProperty( CONFIG_URI ).getValue();
    ...
  }

  @Override public String getProperty( String key ) { return properties.getProperty( key ); }

  public static final PropertyDescriptor CONFIG_URI = new PropertyDescriptor.Builder()
      .name( "Configuration Directory" )
      .description( "Configuration directory for properties files." )
      .defaultValue( null )
      .addValidator( StandardValidators.NON_EMPTY_VALIDATOR )
      .build();
}

Processor.java:

This is the consuming processor. Note that it never references the controller service directly when it identifies the service nor when it gets the product of the service, but always and only the interface upon which the controller service is built:

package com.etretatlogiciels.nifi.processor;

import com.etretatlogiciels.nifi.controller.interfaces.PropertiesFileService;

@CapabilityDescription( "Fetch value from properties service." )
public class Processor extends AbstractProcessor
{
  @Override
  public void onTrigger( final ProcessContext context, final ProcessSession session ) throws ProcessException
  {
    final String                propertyName      = context.getProperty( PROPERTY_NAME ).getValue();
    final PropertiesFileService propertiesService = context.getProperty( PROPERTIES_SERVICE )
                                      .asControllerService( PropertiesFileService.class );
    final String                property          = propertiesService.getProperty( propertyName );
    ...
    session.transfer( flowfile, SUCCESS );
  }

  public static final PropertyDescriptor PROPERTY_NAME = new PropertyDescriptor.Builder()
      .name( "Property Name" )
      .required( true )
      .addValidator( StandardValidators.NON_EMPTY_VALIDATOR )
      .build();
  public static final PropertyDescriptor PROPERTIES_SERVICE = new PropertyDescriptor.Builder()
      .name( "Properties Service" )
      .description( "System properties loader" )
      .required( false )
      .identifiesControllerService( PropertiesFileService.class )
      .build();

  ...
}

Last, but far from least, the JUnit test...

Another challenge, equally difficult to coding a controller service is writing a test for one. The highlighted lines are those germane to testing the controller service and not other testing issues.

ProcessorTest.java:

Notice that, while a new instance of the controller service itself, StandardPropertiesFileService, is created, it's again the interface that's referenced and not the service class itself (line 13). So, when it's time to add the controller service to the test framework (runner) and to enable it, it's the interface reference that's used (line 26).

Note how the controller-service property is loaded ahead of time into a private hash map (line 16) referencing the static PropertyDescriptor in the service (there is no PropertyDescriptor in the interface) directly prior to the association of the properties map with the interface reference when the service is added to the test framework (line 26 as already noted).

Let's detail the three lines (26-28) that are of crucial importance. If you don't get these right, there are some rather cryptic errors that come out of the test framework.

26. addControllerService() —the first argument is free, you can pass any string you like (but you should make it the same as the one passed as the last argument in line 28). The second argument is the interface of the instantiated controller service (from line 13). The third argument is the properties map (list) you contructed earlier. This is the only opportunity you have to establish the service and its properties and their values.
    Important note: if you have optional properties, make sure not to pass nulls. Simply do not include those properties in the list.
27. enableControllerService() —this is just the interface of the instantiated controller service (from line 13).
28. setProperty() —finally, you must inject the property (using its PropertyDescriptor) that is the controller service in the processor that's under test here. Its value is the made-up string from line 26.

The other properties shown explicitly set up in the framework (one, actually, on line 31), using setProperty(), are those of the processor being tested (and not of the controller service).

package com.etretatlogiciels.nifi.processor;

import com.etretatlogiciels.nifi.controller.StandardPropertiesFileService;
import com.etretatlogiciels.nifi.controller.interfaces.PropertiesFileService;

public class ProcessorTest
{

  @Test
  public void testOnTrigger() throws IOException, InitializationException
  {
    TestRunner            runner                      = TestRunners.newTestRunner( new Processor() );
    PropertiesFileService propertiesService           = new StandardPropertiesFileService();
    Map< String, String > propertiesServiceProperties = new HashMap<>();

    propertiesServiceProperties.put( StandardPropertiesFileService.RELOAD_INTERVAL.getName(), "30 sec" );

    URL url = this.getClass().getClassLoader().getResource( "test.properties" );
    assertNotNull( "Should not be null", url );

    String propFile = url.getFile();

    propertiesServiceProperties.put( StandardPropertiesFileService.CONFIG_URI.getName(), propFile );

    // Add controller service
    runner.addControllerService( "propertiesServiceTest", propertiesService, propertiesServiceProperties );
    runner.enableControllerService( propertiesService );
    runner.setProperty( Processor.PROPERTIES_SERVICE, "propertiesServiceTest" );

    // Add properties
    runner.setProperty( Processor.PROPERTY_NAME, "hello" );

    // Add the content to the runner
    runner.enqueue( "TEST".getBytes() );

    // Run the enqueued content, it also takes an int = number of contents queued
    runner.run( 1 );

    // All results were processed with out failure
    runner.assertQueueEmpty();

    // If you need to read or do additional tests on results you can access the content
    List< MockFlowFile > results = runner.getFlowFilesForRelationship( Processor.SUCCESS );
    assertTrue( "1 match", results.size() == 1 );
    MockFlowFile result = results.get( 0 );

    // Test attributes and content
    result.assertAttributeEquals( "property", "nifi.rocks.prop" );
  }
}

Thursday, 18 May 2017

Some notes on ExecuteScript, performance and Jython...

ExecuteScript to run a Python script that runs on a single flowfile despite having multiple flowfiles in the queue is very inefficient. The slowness is due to the Jython initialization time.

At the top of your Python script, put:

flowFiles = session.get( 10 )
for flowFile in flowFiles:
    if flowFile is None:
        continue
    # do stuff here...

Here are two sample scripts.

flowfiles = session.get( 10 )
for flowfile in flowfiles:
    if flowfile is None:
        continue
    s3_bucket = flowfile.getAttribute( 'job.s3_bucket' )
    s3_path   = flowfile.getAttribute( 'job.s3_path' )
    # More stuff here....
    errors = []
    # More stuff here...
    if len( errors ) > 0:
        flowfile = session.putAttribute( flowfile, 'job.error', ';'.join( errors ) )
        session.transfer( flowfile, REL_FAILURE )
    else:
        flowfile = session.putAttribute( flowfile, 'job.number_csv_files', str( len( matches ) ) )
        flowfile = session.putAttribute( flowfile, 'job.total_file_size', str(total_size ) )
        session.transfer( flowfile, REL_SUCCESS
)

import sys
import traceback
from java.nio.charset import StandardCharsets
from org.apache.commons.io import IOUtils
from org.apache.nifi.processor.io import StreamCallback
from org.python.core.util import StringUtil


flowfiles = session.get( 10 )
for flowfile in flowfiles:
    if flowfile is None:
        continue
    start     = int( flowfile.getAttribute( 'range.start' ) )
    stop      = int( flowfile.getAttribute( 'range.stop' ) )
    increment = int( flowfile.getAttribute( 'range.increment' ) )
    for x in range( start, stop + 1, increment ):
        newFlowfile = session.clone( flowfile )
        newFlowfile = session.putAttribute( newFlowfile, 'current', str( x ) )
        session.transfer( newFlowfile, REL_SUCCESS )
        session.remove( flowfile )

Friday, 19 May 2017

NiFi cluster node identifier...

When a node joins a cluster it writes its node identifier into its local state. This is in a state directory under the NiFi installation, which is state/local, I think.

If that subdirectory is removed, the node will get a new identifier. If the subdirectory is left in place, the node gets that same identifier when the NiFi node is started again.

How to process files sequentially...

You have many files in some subdirectory. Let's say they're named file1.csv, file2.csv, etc. How to ensure they're processed sequentially?

There is a processor, EnforceOrder, available beginning in NiFi 1.2.0. Here's a (nice, if complex) flow that Koji Kawamura offers that solves this (and a number of other problems that might arise):

Wednesday, 24 May 2017

Test framework details...

Matt Burgess answered a question from me in the forum and gave some useful detail. I was struggling to do some Mockito work on a class instantiated in init().

A processor's init() method should be getting called at creation time, sic:

TestRunner runner = TestRunners( new MyProcessor() );

It calls Processor.initialize() and AbstractSessionFactoryProcessor's implementation calls its own protected init() method (which is usually what MyProcessor overrides). It also calls any methods that were annotated with @OnAdded and @OnConfigurationRestored. These could be split out if necessary, or organized just as TestRunner's methods are. TestRunner has some overloaded methods for run(), the base one is:

public void run( final int iterations, final boolean stopOnFinish, final boolean initialize, final long runWait );

If you call run() with no parameters, you'll get 1 iteration that initializes then stops on finish (with a 5 second run-wait). However specifying initialize as true will only invoke methods bearing an @OnScheduled annotation.

If you keep a reference to your instance of MyProcessor, you can call your @OnScheduled method explicitly (rather than via TestRunner), then perform your assertions, etc. Then if you want to run onTrigger() but you don't want to reinitialize, you can do:

runner.run( 1, true, false )

...which says to run once, stop on finish, and don't initialize.

Thursday, 1 June 2017

Here's how to interact with a flowfile in both input- and output modes:

flowfile = session.write( flowfile, new StreamCallback()
{
  @Override
  public void process( InputStream in, OutputStream out ) throws IOException
  {
    try
    {
      // use InputStream in and OutputStream out at the same time...
    }
    finally
    {
      in.close();
      out.close();
    }
  }
});

Friday, 2 June 2017

Steps to keep original flowfile (for passing along) while creating a new flowfile with content too.

1. FlowFile original = flowfile;
2. AtomicReference< String > newContentHolder = new AtomicReference<>();
3. Fill new content with:

   session.read( flowfile, new InputStreamCallback()
   {
     @Override
     public void process( InputStream existing ) throws IOException
     {
       newContentHolder.set( ... );
     }
   });
   
   

4. flowfile = session.create( flowfile ); // (keeps attributes)
5. Empty new content into new flowfile with:

   flowfile = session.write( flowfile, new OutputStreamCallback()
   {
     @Override
     public void process( OutputStream out ) throws IOException
     {
       out.write( newContentHolder.getBytes() );
     }
   });
   
   

6. session.transfer( flowfile, SUCCESS );
7. session.transfer( original, SUCCESS );

Unknown relationship at runtime...

You get this exception saying that your relationship is not known:

java.lang.IllegalArgumentException: this relationship relationship-name is not known

Did you forget one of these elements in your processor code?

1. Define relationships (don't use any from some other processor's code):

   public static final Relationship relationship-name = new Relationship.Builder()
       ...
       .build();
   
   

2. Collect relationships into a Set:

   @Override
   public void init( final ProcessorInitializationContext context )
   {
     Set< Relationship > relationships = new HashSet<>();
     relationships.add( relationship-name );
     this.relationships = Collections.unmodifiableSet( relationships );
   }
   
   

3. Make available required method, getRelationships():

   @Override public Set< Relationship > getRelationships() { return relationships; }
   
   

Wednesday, 7 June 2017

Towards tighter control over viewing flowfile content...

Is there a way to know when a flowfile is looked at by someone in the UI?

We have flowfiles containing personal health data (PHI) which no one is supposed to see, but in the case where it's unavoidably crucial to take a look, for debugging or otherwise observing the functioning of a flow, we must know the extent of exposure for legal reasons.

Joe Witt answered,

Thursday, 8 June 2017

NiFi's StateManager...

I need a very light amount of state management, only one variable. However, the basic functionality is to give you Map< String, String > as a map of all the values you want. To get one value, it's:

private String getValueFromStateManager( final ProcessContext context, final String key )
{
  StateMap map = context.getStateManager().getState( Scope.LOCAL );
  return( map != null ) ? map.get( key ) : null;
}

and, to manage the value, you have to juggle the entire map. Here, I replace it wholesale since I've only got one value I care about.

private void putLastTransactionNumber( final ProcessContext context, final String key, final String value )
    throws IOException
{
  Map< String, String > map = new HashMap<>( 1 );
  map.put( key, value );
  context.getStateManager().setState( map, Scope.LOCAL );
}

Bigger example

This isn't a whole custom processor, just as much code as you'd need to recognize what's going on. The state being saved is nothing more than the index of the last operation undertaken successfully.

It looks complicated, but typically, you would handle all of this in small methods rather than right inside your custom processor's onTrigger() method.

State is specific to an instance of a processor on the flow. Safety ends there. If the processor is allowed to have multiple tasks running concurrently then the state mechanism's usage will have to be protected just as with any other variable in the scope of the processor. If the scope is Scope.LOCAL then this is really all you need to think about. If, however, the scope is Scope.CLUSTER then, generally speaking, the intent of usage for that is often associated with a primary node-only action like a ListX processor (like ListFile) with the idea being that the state can be restored by some other node if it's assigned that role.

Friday, 16 June 2017

It seems useful to note, so it turns up in a search, how/why a transfer Relationship might not be specified. I've got to where this isn't hard for me to find, but it's something to note nevertheless for posterity.

Quite often, when a "transfer relationship not specified" error occurs, it's the result of a bug in error handling. For example, the flowfile is to be routed to failure for some reason, but the part of the code that does that has itself a problem and fails.

Wednesday, 21 June 2017

Load-balancing and remote -rocess groups...

Remote-process groups (RPG) handle balancing and failover for you. This is a comment by Bryan Bende.

There's no need to test the RPG; you shouldn't need to use, for example, the DistributeLoad processor. You should be able to have a GenerateFlowFile → RPG (with URL of any node) and then an input port going to some processor. The RPG figures out all of the nodes in the cluster and sends data to all of them.

If you go into the Remote Ports menu of the RPG, each port has some settings that control how many flowfiles get sent per transaction. Generally you will probably get a more even distribution with a smaller batch size, but you will get much better performance with a larger batch size. The RPG also factors in the number of flow files on each node when determining where to send them, so if node 1 is the primary node and has more flow files queued, then the RPG will likely send more flow files to node 2.

Thursday, 29 June 2017

Clustering and Zookeeper...

One reason to run an external Zookeeper in production mode is to prevent trouble arising when Zookeeper cannot establish a "quorum." This will keep a two-node cluster from starting up, for instance.

Keeping in the embedded Zookeeper nevertheless, to start the cluster, remove the server.2 line from zookeeper.properties on node 1; on node 2, set the nifi.state.management.zookeeper.start property (in nifi.properties) to false. Thereafter, as long as node 1 is started first, node 2 will start also.

The scenario above is obviated by an external Zookeeper or by adding a third node to the cluster.

NiFi processors blocking subsequent flowfiles after a failure...

Note that, unless coded to do so, a processor doesn't just stop working because a flowfile running through it happened to produce a failure. Others in the queue behind it will flow when they get their turn.

The processor will retry the failed flowfiles (whose original is left behind in the queue) after a penalty period of 30 seconds. This value is configurable in the processor's Settings → Penalty Duration.

Meanwhile, other, "innocent" files continue to make their way through the processor. The processor isn't summarily halted unless it's coded to halt at first failure.

Tuesday, 4 July 2017

The new NiFi registry...

Reminder...

Wednesday, 5 July 2017

Just a note...

NiFi's conf subdirectory is not on any processor's classpath.

Friday, 7 July 2017

NiFi CLASSPATH...

The conf subdirectory is on the class path of the system class loader when NiFi is launched. The class hierarchy is something like this:

1. System class loader (conf subdirectory and JAR files in the lib subdirectory)
2. Jetty NAR class loader (parent is sytem class loader)
3. (for instance:) Geomesa NAR class loader (parent is Jetty class loader)

The classpath of the system class load should be what you see in logs/nifi-bootstrap.log for the command that launched NiFi.

Wednesday, 19 July 2017

NiFi behind a reverse proxy (webserver)

How to set up httpd/conf.d/nifi.conf (or /etc/apache2/sites-available/nifi.conf to run behind a reverse proxy server:

ProxyPass        / http://hostname or address:8080/
ProxyPassReverse / http:/hostname or address:8080/

RequestHeader add X-ProxyScheme      "http"
RequestHeader add X-ProxyHost        "hostname or address"
RequestHeader add X-ProxyPort        "80"
RequestHeader add X-ProxyContextPath "/"

Thursday, 20 July 2017

NiFi support for large data sets...

From Joe Witt...

NiFi can handle very small and very large datasets very well today. It has always been the case with NiFi that it reads data from its content repository using input streams and writes content to its content repository using output streams. Because of this, developers building processors can design components which pull in 1-byte objects just as easily as it can pull in 1Gb objects (or much more). The content does not ever have to be held in memory in whole except in those cases where a given component is not coded to take advantage of this.

Consider a process which pulls data via SFTP or HTTP or some related protocol. NiFi pull data from such an endpoint is streaming the data over the network and writing it directly to disk. We're never holding that whole object in some byte[] in memory. Similarly, when large objects move from processor to processor we're just moving pointers.

In fact, NiFi can even handle cases like merging content together very well for the same reason. We can accumulate a bunch of source content/flowfiles into a single massive output and do so having never held it all in memory at once.

Now, where can memory trouble happen? It can happen when the number of flowfile objects (the metadata about them—not the content) are held in memory at once. These can accumulate substantially in certain cases like splitting data for example. Consider an input CSV file with 1,000,000 lines. One might want individual lines so they can run specific per event processes on it. This can be accomplished in a couple ways.

1. They can use SplitText to split into single lines. We'll almost surely run out of heap since we're making 1,000,000 flowfile objects. Avoid this option.
2. They can use SplitText first splitting into say 1,000 lines at a time. This will produce 1,000 output flowfiles each with 1,000 lines in it. They can then do another split text which splits to single lines. This way no single session will ever have more than 1,000 splits in it and things will work great. This combined with backpressure and provenance works extremely well.
3. They can use the new record reader/writer processors which support really powerful and common patterns in a format and schema aware manner and which makes setup really easy and reusable. In fact, this approach in many cases means they dont even need to split up the data since the record-oriented processors will know how to demarcate events "as they exist in their source form." Huge performance gains and usability improvement here.

Monday, 24 July 2017

Kafka

Been looking at Kafka of late. Other than a missing, nice user interface like NiFi's, what is the real distinction? The one is message-queueing (Kafka) and the other a flow-control framework (NiFi).

If NiFi is providing all that's needed then Kafka isn't prescribed. However, Kafka offers an important capability in a broad, enterprise-requirement sense, the provision of a highly durable and replayable buffer of insertion ordered data for efficient access.

Wednesday, 26 July 2017

Jetty failure to start: org.jasypt.exceptions.EncryptionOperationNotPossibleException

We saw this once:

2017-07-25 23:23:31,148 WARN [main] org.apache.nifi.web.server.JettyServer
Failed to start web server... shutting down.
org.apache.nifi.encrypt.EncryptionException:
org.jasypt.exceptions.EncryptionOperationNotPossibleException
    at
org.apache.nifi.encrypt.StringEncryptor.decrypt(StringEncryptor.java:149)
~[nifi-framework-core-1.1.2.jar:1.1.2]
    at
org.apache.nifi.controller.serialization.FlowFromDOMFactory.decrypt(FlowFromDOMFactory.java:474)
~[nifi-framework-core-1.1.2.jar:1.1.2]
.
.
.

The problem solution was found based on a quickly answered suggestion by Mark Payne:

Our Ansible instructions upgraded NiFi and created a new nifi.sensitive.props.key. In nifi.properties this property, if extant, is used to encrypt sensitive properties in flow.xml.gz. Thus, upon relaunching NiFi, the wrong key was used to decrypt resulting in the reported failure to start, flow.xml.gz is no longer useful.

How did we solve it?

We looked in the nifi.properties.rpmsave file, what RPM does with a file it's changed, and copied the old key from this property to paste in over the newly generated key in nifi.properties. Relaunched, NiFi worked with no problem. The full solution, in our case, is to insist in Ansible that it not generate for and replace nifi.sensitive.props.key with a new key.

Tuesday, 19 September 2017

Peter Wicks wrote this in the forum. I decided to copy it here for safe-keeping.

Test 1: 1 Concurrent, non thread-safe

1. The purpose of this test is to find out what happens to a processor's state between execution of flowfiles.
2. After 10,000 files the value was 10,000 on the last file. This means that state is maintained in a processor between runs (this was what I assumed, but a good place to start). Each execution of the processor used the same instance of the class.

Test 2: Stop the processor, then start it again and run a single file

1. The purpose of this test is to figure out when a processor resets its state.
2. After 1 file the value was 10,001.
3. This means that stopping and starting a processor does not reset the processor. The same class instance is still used.

Test 3: Stop, disable, enable, and then start again.

1. The purpose of this test is to see if disabling a processor causes the class instance to be disposed of.
2. After 1 file the value was 10,002.
3. This means that disabling and re-enabling a processor does not reset its state. The same class instance persists.

Test 4: Starting with a new copy of the test processor, run 10 concurrent threads, non thread-safe

1. The purpose of this test is to see if each thread uses its own instance of the class, or if a shared instance of the class is used.
2. After 10,000 files, the value was 9,975 on the last file. I didn't run this test more than once, but the value should fluctuate from run to run due to thread contention.
3. I saw at least 8 concurrent threads running at one point. Combined with this, and the resulting value, I'm fairly confident that the same class instance is used for all concurrent threads.

Test 5: Starting with a new copy of the test processor, run 1 concurrent thread, thread-safe

1. The purpose of this test is to find out what happens to a processor's state between execution of flowfiles.
2. After 10,000 files the value was 10,000 on the last file.
3. This means that state is maintained in a processor between runs (this matches the non-thread safe results, which makes sense for 1 concurrent thread).

Test 6: Starting with a new copy of the test processor, run 10 concurrent threads, thread-safe

1. The purpose of this test is to contrast with the non thread-safe approach, and verify that a thread-safe object will work across concurrent threads.
2. After 10,000 files the value was 10,000 on the last file.
3. This means that thread synchronization works with multiple, concurrent threads for a single class instance.

These tests ran on a single NiFi instance, with no clustering and are not designed to say anything about clustering.

Based upon my limited test results, a processor class is never reinstantiated unless the processor is deleted from the flow (... yea, kind of like cheating) or NiFi restarts.

If you have multiple versions of the same processor loaded into your flow (applicable only to 1.3, or perhaps 1.2?), you will also see the processor class reinstantiated when you right-click on a processor and choose to change the version.

Monday, 25 September 2017

According to Joe Witt, the most common sources of memory leaks in custom processors are

1. loading large objects (contents of flowfiles, for example) into memory through making them byte[] or, doing so using libraries that do this and not realizing that these libraries are doing this,
2. doing the above in parallel, that is, multiple flowfiles at once,
3. caching objects in memory failing to provide bounds on this action,
4. failing to size the JVM heap appropriate to flow,
5. pulling in lots of flowfiles into a single session, or...
6. creating many flowfiles in a single session.

Tuesday, 26 September 2017

Jow Witt on NiFi site-to-site:

1. "Does NiFi site-to-site scale well? Are there any performance concerns or limitations I need to consider? Can I send live traffic from thousands of NiFi clusters to a single NiFi cluster (let's suppose the central one is a huge cluster)? Is there any limitation on the number of input ports for example?"
   
   

   There are practical limits on how many input ports there can be. Each port generates threads to manage those sockets. However, many different edge systems can send to a single input port. You can also demultiplex the streams of data using flowfile attributes, so there are various ways to tackle that. It wasn't tested against thousands of edge systems sending to a central cluster as the more common model in such a case is less of tons of spokes and one central hub rather with spokes sending to regional clusters which send to central cluster(s).

2. "How can NiFi handle load-balancing in the site-to-site situation? Is that per session or flow or in a batch mode? I want to understand all the situations we may face regarding the network issues between different NiFi clusters."
   
   

   Site-to-site has load balancing and fail-over built into it. The site-to-site exchange that happens when the connection is established and over time is to share information about the cluster, how many nodes are in it and their relative load. This allows the clients to do weighted distribution, detect new or removed nodes, etc.

3. "Do I need to use the same version of NiFi from the edge NiFi clusters to the central NiFi clusters? Is there any chance that site-to-site communication will change significantly such that we need to upgrade all the NiFi instances we will have or it is pretty reliable now?"
   
   

   You don't have to use the same version. This is another benefit of site-to-site is that it was built with the recognition that it is not possible or even desirable to upgrade all systems at once across a large enterprise. The protocol involves both sides of site-to-site transfers to exchange information about the flowfile/transfer protocol it supports. So old NiFi sending to new NiFi and new NiFi sending to an old NiFi are able to come to basic agreement. The protocol and serialization have been quite stable but still the ability to evolve is baked in.

Wednesday, 27 September 2017

To get ComponentLog output to stdout during unit testing of custom processors, you can configure this in src/test/resources/logback.xml, likely not already present. However, you also need to add

<dependency>
  <groupId>ch.qos.logback</groupId>
  <artifactId>logback-classic</artifactId>
  <version>1.2.3</version>
  <scope>test</scope>
</dependency>

Contents for src/test/resources/logback.xml might look like this:

<configuration>
  <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
    <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
      <pattern>%-4r [%t] %-5p %c - %m%n</pattern>
    </encoder>
  </appender>

  <appender name="FILE" class="ch.qos.logback.core.FileAppender">
    <file>./target/log</file>
    <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
      <pattern>%date %level [%thread] %logger{40} %msg%n</pattern>
    </encoder>
  </appender>

  <logger name="org.apache.nifi" level="INFO" />
  <root level="INFO">
    <appender-ref ref="CONSOLE" />
  </root>
</configuration>

Wednesday, 4 October 2017

As soon as it showed up in the downloads, I grabbed NiFi 1.4.0 to check it out. One of the new, cool features is variables. There's no documentation yet, but I reached them by right-clicking on a process group, choosing Variables, then adding one just for fun. My process group is named Poop:

NiFi variables can be referenced from situations in which the NiFi Expression Language is being used. It appears to me that, missing a variable at one level (process group), NiFi will crawl up the hierarchy (of process groups) until it finds it. This is in addition to NiFi's environment variable (registry) capability which can also be referenced from the NiFi Expression Language.

Tuesday, 17 October 2017

Working example of NiFi local state management

In this example, we have a list of log files that we're reading periodically one line at a time. We want to keep our place in each file, that is, the last line we've already read. We key this information off the logfile path, which is discrete no matter how many different ones there are since these are different pathnames in the filesystem.

/**
 * Return the last line processed for the given file in order to be able to
 * start on the line after this one.
 * @param context current NiFi processor context.
 * @param pathname of log file.
 * @throws IOException in case of some disappointment.
 */
protected long getLastLineProcessed( final ProcessContext context, final String pathname )
    throws IOException
{
  StateMap localState = context.getStateManager().getState( Scope.LOCAL );
  String   lastLine   = localState.get( pathname );
  return( lastLine != null ) ? Long.parseLong( lastLine ) : 0L;
}

/**
 * Save last line processed for the given file.
 * @param context current NiFi processor context.
 * @param pathname of log file.
 * @param line last line processed: next time, start after this point.
 * @throws IOException in case of some disappointment.
 */
protected void saveLastLineProcessed( final ProcessContext context, final String pathname, long line )
    throws IOException
{
  StateMap              localState       = context.getStateManager().getState( Scope.LOCAL );
  Map< String, String > copyOfLocalState = localState.toMap();
  int                   stateSize        = ( copyOfLocalState.size() > 0 ) ? copyOfLocalState.size() : 1;
  Map< String, String > newState         = new HashMap<>( stateSize );
  newState.putAll( copyOfLocalState );
  newState.put( pathname, line+"" );
  context.getStateManager().setState( newState, Scope.LOCAL );
}

How it works:

The state is held as a single map. In that map, there is information, each a string, each keyed by some string. The map can only replaced as a whole, so get it, manage the whole, and replace the whole.

Friday, 3 November 2017

Someone asserted that, whenever the number of processors in their flow exceeded something like 1500, the whole instance of NiFi slowed down noticeably and the UI became unresponsive.

Michael Moser wrote back pointing out that any time processors find themselves in the STOPPED state, NiFi is nevertheless performing validation on them including all their controller services. This would have the effect of slowing the UI.

A better solution to this problem is to put the unused processors into the DISABLED state as NiFi skips validation of processors in this or the RUNNING states.

A disabled processor appears thus:

It's put into this state by unclicking the Enabled checkbox:

Thursday, 9 November 2017

IT is having trouble with NiFi on a server. There are bunch of files under work that don't exist. Here's NiFi 0.7.4 installation before running. Notice that there's no logs nor work subdirectory:

russ@nargothrond ~/dev/nifi $ ll nifi-0.7.4/
total 156
drwxrwxr-x 6 russ russ  4096 Jun  5 15:28 .
drwxr-xr-x 6 russ russ  4096 Nov  9 08:56 ..
drwxrwxr-x 2 russ russ  4096 Jun  5 15:28 bin
drwxrwxr-x 2 russ russ  4096 Jun  5 15:28 conf
drwxrwxr-x 3 russ russ  4096 Jun  5 15:28 docs
drwxrwx--- 3 russ russ  4096 Jun  5 15:28 lib
-rw-r--r-- 1 russ russ 79054 Jun  5 14:31 LICENSE
-rw-r--r-- 1 russ russ 42601 Jun  5 14:31 NOTICE
-rw-r--r-- 1 russ russ  4549 Jun  5 14:31 README

After launching, these files exist...

russ@nargothrond ~/dev/nifi/nifi-0.7.4 $ ./bin/nifi.sh start

Java home: /home/russ/dev/jdk1.8.0_144
NiFi home: /home/russ/dev/nifi/nifi-0.7.4

Bootstrap Config File: /home/russ/dev/nifi/nifi-0.7.4/conf/bootstrap.conf

2017-11-09 08:59:48,822 INFO [main] org.apache.nifi.bootstrap.Command Starting Apache NiFi...
2017-11-09 08:59:48,822 INFO [main] org.apache.nifi.bootstrap.Command Working Directory: /home/russ/dev/nifi/nifi-0.7.4
2017-11-09 08:59:48,822 INFO [main] org.apache.nifi.bootstrap.Command Command: /home/russ/dev/jdk1.8.0_144/bin/java
  -classpath /home/russ/dev/nifi/nifi-0.7.4/./conf
    :/home/russ/dev/nifi/nifi-0.7.4/./lib/nifi-runtime-0.7.4.jar
    :/home/russ/dev/nifi/nifi-0.7.4/./lib/nifi-documentation-0.7.4.jar
    :/home/russ/dev/nifi/nifi-0.7.4/./lib/slf4j-api-1.7.12.jar
    :/home/russ/dev/nifi/nifi-0.7.4/./lib/nifi-api-0.7.4.jar
    :/home/russ/dev/nifi/nifi-0.7.4/./lib/jcl-over-slf4j-1.7.12.jar
    :/home/russ/dev/nifi/nifi-0.7.4/./lib/log4j-over-slf4j-1.7.12.jar
    :/home/russ/dev/nifi/nifi-0.7.4/./lib/jul-to-slf4j-1.7.12.jar
    :/home/russ/dev/nifi/nifi-0.7.4/./lib/logback-core-1.1.3.jar
    :/home/russ/dev/nifi/nifi-0.7.4/./lib/logback-classic-1.1.3.jar
    :/home/russ/dev/nifi/nifi-0.7.4/./lib/nifi-nar-utils-0.7.4.jar
    :/home/russ/dev/nifi/nifi-0.7.4/./lib/nifi-properties-0.7.4.jar
  -Dorg.apache.jasper.compiler.disablejsr199=true -Xmx512m -Xms512m
    -Dsun.net.http.allowRestrictedHeaders=true
    -Djava.net.preferIPv4Stack=true
    -Djava.awt.headless=true
    -Djava.protocol.handler.pkgs=sun.net.www.protocol
    -Dnifi.properties.file.path=/home/russ/dev/nifi/nifi-0.7.4/./conf/nifi.properties
    -Dnifi.bootstrap.listen.port=36755
    -Dapp=NiFi
    -Dorg.apache.nifi.bootstrap.config.log.dir=/home/russ/dev/nifi/nifi-0.7.4/logs org.apache.nifi.NiFi
russ@nargothrond ~/dev/nifi/nifi-0.7.4 $ ll work/jetty/
total 44
drwxrwxr-x 11 russ russ 4096 Nov  9 08:59 .
drwxrwxr-x  5 russ russ 4096 Nov  9 08:59 ..
drwxrwxr-x  4 russ russ 4096 Nov  9 08:59 nifi-image-viewer-0.7.4.war
drwxrwxr-x  4 russ russ 4096 Nov  9 08:59 nifi-jolt-transform-json-ui-0.7.4.war
drwxrwxr-x  4 russ russ 4096 Nov  9 08:59 nifi-standard-content-viewer-0.7.4.war
drwxrwxr-x  4 russ russ 4096 Nov  9 08:59 nifi-update-attribute-ui-0.7.4.war
drwxrwxr-x  4 russ russ 4096 Nov  9 08:59 nifi-web-api-0.7.4.war
drwxrwxr-x  4 russ russ 4096 Nov  9 08:59 nifi-web-content-viewer-0.7.4.war
drwxrwxr-x  4 russ russ 4096 Nov  9 08:59 nifi-web-docs-0.7.4.war
drwxrwxr-x  4 russ russ 4096 Nov  9 08:59 nifi-web-error-0.7.4.war
drwxrwxr-x  4 russ russ 4096 Nov  9 08:59 nifi-web-ui-0.7.4.war

After halting, these files disappear...

russ@nargothrond ~/dev/nifi/nifi-0.7.4 $ ./bin/nifi.sh stop

Java home: /home/russ/dev/jdk1.8.0_144
NiFi home: /home/russ/dev/nifi/nifi-0.7.4

Bootstrap Config File: /home/russ/dev/nifi/nifi-0.7.4/conf/bootstrap.conf

2017-11-09 09:01:47,148 INFO [main] org.apache.nifi.bootstrap.Command Apache NiFi has accepted the Shutdown Command and is shutting down now
2017-11-09 09:01:47,220 INFO [main] org.apache.nifi.bootstrap.Command NiFi has finished shutting down.
russ@nargothrond ~/dev/nifi/nifi-0.7.4 $ ll work/jetty/
total 8
drwxrwxr-x 2 russ russ 4096 Nov  9 09:01 .
drwxrwxr-x 5 russ russ 4096 Nov  9 08:59 ..

This is happening because NiFi is failing to start up after some damage IT's synchronization script has done to the installation. I'm guessing it's the flow.xml.gz file, but they're sending me a scrap of