Notes on writing a simple NiFi custom processor

 
  session.transfer( newFlowfile, SUCCESS );
}

 
  session.transfer( newFlowfile, SUCCESS );
  session.remove( flowfile );
}

 
  session.transfer( newFlowfile, NEW );
  session.transfer( flowfile, OLD );
}

 
  flowfile = session.putAttribute( flowfile, "attribute-name", "value" );
 

 
  flowfile = session.putAttribute( flowfile, "attribute-first", "value" );
  session.putAttribute( flowfile, "attribute-second", "value" );
  .
  .
  .
  session.transfer( flowfile, SUCCESS );
}

package com.windofkeltia.processor;

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

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.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.util.StandardValidators;

/**
 * Sample Monopoly-themed custom processor paired with simple JUnit test.
 * This processor is consciously for beginning to learn how to write and
 * test a custom processor.
 * @author Russell Bateman
 * @since July 2020
 */
@SideEffectFree
@Tags( { "instruction" } )
@CapabilityDescription( "Illustrates beginning to write custom processors." )
public class Monopoly extends AbstractProcessor
{
  @Override public void onTrigger( ProcessContext context, ProcessSession session ) throws ProcessException
  {
    FlowFile flowfile = session.get();

    if( flowfile == null )
    {
      context.yield();
      return;
    }

    /* ------------------------------------------------------------------------------------
     * Here's where the real work of a custom processor goes, usually containing a call to:
     * a) session.read( flowfile, new InputStreamCallback()...
     * b) session.write( flowfile, new OutputStreamCallback()... or
     * c) session.write( flowfile, new StreamCallback()...
     *
     * The original flowfile is read, maybe a new one is written to output.
     * ------------------------------------------------------------------------------------
     */

    // how we reach nifi-app.log...
    getLogger().info( "Just passing through on our way to Go..." );

    // how to get the value of a property...
    String favorite = context.getProperty( MONOPOLY_PROPERTY.getName() ).getValue();

    if( !favorite.equals( "Bordwalk" ) )
    {
      session.transfer( flowfile, FAILURE );
      return;
    }

    // how to create and set a new flowfile attribute...
    flowfile = session.putAttribute( flowfile, "property", "Bordwalk" );
    flowfile = session.putAttribute( flowfile, "content", "flowfile unchanged" );
    session.transfer( flowfile, SUCCESS );
  }

  public static final PropertyDescriptor MONOPOLY_PROPERTY = new PropertyDescriptor.Builder()
      .name( "Favorite property name" )           // must always remain the same or flows using this processor will break!
      .displayName( "Nom de propriété préférée" ) // this name can change without breaking flow.xml.gz
      .addValidator( StandardValidators.NON_EMPTY_VALIDATOR )
      .defaultValue( "Bordwalk" )
      .description( "Favorite property name on the Monopoly board." )
      .build();

  public static final Relationship SUCCESS = new Relationship.Builder()
      .name( "Success" )
      .description( "The flowfile passes go." )
      .build();
  public static final Relationship FAILURE = new Relationship.Builder()
      .name( "Failure" )
      .description( "The flowfile doesn't pass go." )
      .build();

  private List< PropertyDescriptor > properties;
  private Set< Relationship >        relationships;

  @Override
  public void init( final ProcessorInitializationContext context )
  {
    List< PropertyDescriptor > properties = new ArrayList<>();
    properties.add( MONOPOLY_PROPERTY );
    this.properties = Collections.unmodifiableList( properties );

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

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

package com.windofkeltia.processor;

import java.io.ByteArrayInputStream;
import java.util.List;
import java.util.Map;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import static org.junit.Assert.assertEquals;
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.windofkeltia.utilities.StringUtilities;

/**
 * Set breakpoints in Monopoly.onTrigger() and step through these tests
 * in the debugger.
 * @author Russell Bateman
 * @since July 2020
 */
public class MonopolyTest
{
  @After public void tearDown() { }
  @Before public void setUp() { runner = TestRunners.newTestRunner( processor ); }

  private final Monopoly   processor = new Monopoly();
  private       TestRunner runner;

  private static final boolean VERBOSE = true;

  @Test
  public void testSuccess()
  {
    System.out.println( "\n--- testSuccess() -----------------------------------------------------------------------" );

    // how to set a property value...
    runner.setProperty( Monopoly.MONOPOLY_PROPERTY, "Bordwalk" );
    // how to create flowfile content...
    runner.enqueue( new ByteArrayInputStream( "Take a ride on the Reading Railroad!".getBytes() ) );
    // this runs the processor; your first breakpoint opportunity is likely onTrigger()...
    runner.run( 1 );
    runner.assertQueueEmpty();

    // get all the flowfiles that were transferred to SUCCESS...
    List< MockFlowFile > flowfiles = runner.getFlowFilesForRelationship( Monopoly.SUCCESS );
    assertEquals(1, flowfiles.size() );

    // we know there's only one flowfile, so get it for our test...
    MockFlowFile          flowfile   = flowfiles.get( 0 );  assertNotNull( flowfile );
    String                content    = new String( runner.getContentAsByteArray( flowfile ) );
    Map< String, String > attributes = flowfile.getAttributes();

    if( VERBOSE )
    {
      displayAttributes( attributes );
      System.out.println( "Content: ---------------------------------------------------------------------------------");
      System.out.println( "  " + content );
    }
  }

  @Test
  public void testFailure()
  {
    System.out.println( "\n--- testFailure() -----------------------------------------------------------------------" );
    runner.setProperty( Monopoly.MONOPOLY_PROPERTY, "Park Place" );
    runner.enqueue( new ByteArrayInputStream( "Go to jail; go directly to jail!".getBytes() ) );
    runner.run( 1 );
    runner.assertQueueEmpty();

    // get all the flowfiles that were transferred to FAILURE...
    List< MockFlowFile > flowfiles = runner.getFlowFilesForRelationship( Monopoly.FAILURE );
    assertEquals(1, flowfiles.size() );

    MockFlowFile          flowfile   = flowfiles.get( 0 );  assertNotNull( flowfile );
    String                content    = new String( runner.getContentAsByteArray( flowfile ) );
    Map< String, String > attributes = flowfile.getAttributes();

    if( VERBOSE )
    {
      displayAttributes( attributes );
      System.out.println( "Content: ---------------------------------------------------------------------------------");
      System.out.println( "  " + content );
    }
  }

  private void displayAttributes( Map< String, String > attributes )
  {
    System.out.println( "Attributes: --------------------------------------------------------------------------------");
    int maxWidth = 0;
    for( Map.Entry< String, String > attribute : attributes.entrySet() )
      maxWidth = Math.max( maxWidth, attribute.getKey().length() );

    for( Map.Entry< String, String > attribute : attributes.entrySet() )
      System.out.println( "  " + StringUtilities.padStringLeft( attribute.getKey(), 9 )
          + ": " + attribute.getValue() );
  }
}

<configuration>
  <!-- This makes the NiFi component logger spit stuff out to the console in running tests.
       We set the level to DEBUG (or TRACE if we use that) to see the maximum.
    -->
  <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
    <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
      <!-- <pattern>%-4r [%t] %-5p %c.%m%n:%L</pattern>-->
      <pattern>%-4relative [%thread] %-5level %logger{5}.%method:%line - %message%n</pattern>
      <!-- %relative outputs the number of milliseconds elapsed since beginning of execution
           %logger{n] where {n} is the abbreviation (number) of or referring to a "scheme", see
           http://logback.qos.ch/manual/layouts.html
        -->
    </encoder>
  </appender>

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

  <logger name="com.windofkeltia.processor" level="DEBUG" additivity="false">
    <appender-ref ref="CONSOLE" />
  </logger>

  <root level="INFO">
    <appender-ref ref="CONSOLE" />
  </root>
</configuration>

package com.windofkeltia.processor;

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

import static java.util.Objects.isNull;

import org.apache.nifi.annotation.behavior.SideEffectFree;
import org.apache.nifi.annotation.behavior.TriggerWhenEmpty;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.logging.ComponentLog;
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.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;

@TriggerWhenEmpty
@SideEffectFree
@CapabilityDescription( "Dynamic properties can be created to specify (or add to) static configuration"
                + " of key-value substitution pairs.")
public class ConfigureProcessor extends AbstractProcessor
{
  @Override
  public void onTrigger( final ProcessContext context, final ProcessSession session ) throws ProcessException
  {
    ComponentLog logger = getLogger();

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

      if( !property.isDynamic() )
        continue;

      // do work here...
      final String PROPERTY_NAME  = property.getName();
      final String PROPERTY_VALUE = entry.getValue();

      logger.trace( "Processing configurable mappings titled \"" + PROPERTY_NAME + "\"" );

      try
      {
        harvestDynamicPropertyMappings( PROPERTY_VALUE );
      }
      catch( Exception e )
      {
        logger.debug( e.getMessage() );
      }
    }
  }

  protected static void harvestDynamicPropertyMappings( final String PROPERTY_VALUE ) throws Exception
  {
    final String[] LINES      = PROPERTY_VALUE.split( "\n" );
    int            lineNumber = 0;

    if( LINES.length < 1 )
      return;

    final String WHICH_LIST = LINES[ 0 ];

    for( int which = 1; which < LINES.length; which++ )
    {
      final String VALUE_LINE = LINES[ which ];
      char         delimiter  = VALUE_LINE.charAt( 0 );
      int          position   = VALUE_LINE.indexOf( delimiter, 1 );
      String key, value;

      key   = ( position < 0 ) ? VALUE_LINE.substring( 1 )                   : VALUE_LINE.substring( 1, position ).trim();
      value = ( position > 0 ) ? VALUE_LINE.substring( position + 1 ).trim() : "";

      switch( WHICH_LIST )
      {
        case "human-readables" :
          HumanReadables.add( key, value );  // (I don't show this stuff...)
          break;
        case "code-system urs" :
          CodeSystemUri.add( key, value );   // (...left up to the imagination)
          break;
        default :
          throw new Exception( "Configurable-mapping " + WHICH_LIST + " is not a supported list" );
      }
    }
  }

  /**
   * This method is overridden from AbstractConfigurableComponent.
   *
   * In allowing a user to define a dynamic property, which is to say
   * one that is not known statically (as are most processor properties), we
   * override this method to create that property allowing us also to specify
   * a validator. Without doing this, we'll always get something like:
   *
   * java.lang.AssertionError: Processor has 1 validation failures:
   * (dynamic-property-name) validated against (its value) is invalid because\
   *   (dynamic-property-name) is not a supported property or has no Validator associated with it
   *
   * @param propertyDescriptorName the name of the property's descriptor.
   * @return the new descriptor.
   */
  @Override
  protected PropertyDescriptor getSupportedDynamicPropertyDescriptor( final String propertyDescriptorName )
  {
    return new PropertyDescriptor.Builder()
                          .required( false )
                          .name( propertyDescriptorName )
                          .addValidator( StandardValidators.NON_EMPTY_VALIDATOR )
                          // or .addValidator( Validator.VALID ) if you do not wish it validated!
                          .dynamic( true )
                          .build();
  }

  /**
   * Used with onPropertyModified(), it collects dynamic properties
   * that change. This is not common.
   * See {@link #onPropertyModified( PropertyDescriptor, String, String )}
   */
  private volatile Set< String > dynamicPropertyNames = new HashSet<>();

  /**
   * This method is overridden from AbstractConfigurableComponent. It's
   * called in reaction to a configuration change (in the UI).
   *
   * Why might you want to do this? Because you are using dynamic properties to
   * influence the set of relationships that a processor may have, adding to or
   * subtracting from them. For example, if we were called with "dynamic-property"
   * added, then this method would add a new relationship (to SUCCESS)
   * called "dynamic-property" (--for whatever good that might do).
   *
   * This is not a common thing to do.
   *
   * @param descriptor of the modified property.
   * @param oldValue non-null: previous property value.
   * @param newValue new property value or null when property removed (trash can in UI).
   */
  @Override
  public void onPropertyModified( final PropertyDescriptor descriptor, final String oldValue, final String newValue )
  {
    final Set< String > newDynamicPropertyNames = new HashSet<>( dynamicPropertyNames );

    if( isNull( newValue ) )
      newDynamicPropertyNames.remove( descriptor.getName() );
    else if( isNull( oldValue ) && descriptor.isDynamic() )
      newDynamicPropertyNames.add( descriptor.getName() );

    dynamicPropertyNames = Collections.unmodifiableSet( newDynamicPropertyNames );

    final Set< String > allDynamicProperties = dynamicPropertyNames;
  }

  /**
   * When the processor is scheduled, you can access its properties including
   * any dynamic ones.
   * @param context processor context.
   */
  @OnScheduled public void processProperties( final ProcessContext context )
  {
    for( Map.Entry< PropertyDescriptor, String > entry : context.getProperties().entrySet() )
    {
      PropertyDescriptor descriptor = entry.getKey();

      if( descriptor.isDynamic() )
        getLogger().debug( "Dynamic property named:\n    " + descriptor.getName()
                     + ", value: " + entry.getValue().replaceAll( "\n", " + " ) );
    }
  }

  public static final PropertyDescriptor ERASE_HUMANREADABLES = new PropertyDescriptor.Builder()
      .name( "Erase all human-readables" )
      .displayName( "Erase all human-readables" )
      .description( "Erase all human-readables and replace with the dynamic ones defined by this processor."
               + "(Medical code-system substitutions, a different list, cannot be erased." )
      .required( false )
      .addValidator( BooleanProperty.BOOLEAN_VALIDATOR )
      .allowableValues( BooleanProperty.BOOLEAN_VALUES )
      .defaultValue( BooleanProperty.FALSE )
      .build();

  private List< PropertyDescriptor > properties;

  @Override
  public void init( final ProcessorInitializationContext context )
  {
    // create static properties and relationships...
    @SuppressWarnings( "MismatchedQueryAndUpdateOfCollection" )
    List< PropertyDescriptor > properties = new ArrayList<>();
    properties.add( ERASE_HUMANREADABLES );
    this.properties = Collections.unmodifiableList( properties );
  }

  @Override public List< PropertyDescriptor > getSupportedPropertyDescriptors() { return properties; }
}

package com.windofkeltia.com.processor;

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

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 org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;

import com.windofkeltia.constants.ConfigurableMappings;

/**
 * @author Russell Bateman
 * @since December 2020
 */
public class ConfigureProcessorTest
{
  // @formatter:off
  @Rule   public TestName name = new TestName();
  @After  public void tearDown() { }
  @Before public void setUp()    { }

  private static final boolean VERBOSE  = true;
  private static final int     ONE      = 1;

  private ConfigureProcessor processor = new ConfigureProcessor();
  private TestRunner         runner    = TestRunners.newTestRunner( processor );

  private static final String PROPERTY_VALUE = "human-readables\n"
                                             + "|this|that\n"
                                             + "|and| everything\n"
                                             + "|nothing\n"
                                             + "^J'accuse^de rien !"
                                             + "\n";
  @Test
  public void test()
  {
    runner.setProperty( "my human readables", PROPERTY_VALUE );
    runner.setValidateExpressionUsage( false );
    runner.run( ONE );
    runner.assertQueueEmpty();

    int line = 0;

    List< String >        output  = new ArrayList<>();
    Map< String, String > mappings = ConfigurableMappings.HumanReadables.getMap();
    for( Map.Entry< String, String > map : mappings.entrySet() )
    {
      String keyAndValue = map.getKey() + " --> " + map.getValue();
      output.add( keyAndValue );
      System.out.println( "  " + ++line + ": " + keyAndValue );
    }

    assertTrue( output.contains( "http://hospital.smarthealthit.org --> Smart Health IT" ) );
    assertTrue( output.contains( "http://diameterhealth.com --> Diameter Health" ) );
    assertTrue( output.contains( "this --> that" ) );
    assertTrue( output.contains( "and --> everything" ) );
    assertTrue( output.contains( "nothing --> " ) );
    assertTrue( output.contains( "J'accuse --> de rien !" ) );
  }

  @Test
  public void testHarvestDynamicPropertyMappings()
  {
    try
    {
      int lineNumber = 0;

      ConfigureProcessor.harvestDynamicPropertyMappings( PROPERTY_VALUE );

      Map< String, String > mappings = HumanReadables.getMap();
      for( Map.Entry< String, String > map : mappings.entrySet() )
      {
        String key   = map.getKey();
        String value = map.getValue();

        if( VERBOSE )
          System.out.println( "  " + ++lineNumber + ": key = " + key + ", value = " + value );
      }
    }
    catch( Exception e )
    {
      if( VERBOSE )
        System.out.println( e.getMessage() );
    }
  }
}

/**
 * Slurp content from input stream and write it to the output stream.
 */
public static void slurpIntoOutputStream( final InputStream inputStream,
                               final OutputStream outputStream,
                               int bufferSize )
    throws IOException
{
  if( isNull( inputStream ) )
    throw new IOException( "Missing input stream" );
  else if( isNull( outputStream ) )
    throw new IOException( "Missing output stream" );

  byte[] buffer = new byte[ bufferSize ];

  for( ; ; )
  {
    int length = inputStream.read( buffer, 0, bufferSize );
    outputStream.write( buffer, 0, length );

    if( length < bufferSize )
      break;
  }
}