How to use context parameters in a java custom processor and unit test

Question

I have created a trial Nifi processor in Java with 2 property descriptors which are intended to each read a context parameter

   public static final PropertyDescriptor
  CACHE_KEY =
  property( "last-dt-processed-key" )
    .description( "References a context property which holds the cache key used for the last date time processed" )
    .required( true )
    .addValidator( NON_EMPTY_VALIDATOR )
    .defaultValue( "#{last-dt-processed-key}" )
    .build( );

public static final PropertyDescriptor
  PROCESS_FROM_DAYS_AGO_DEFAULT =
  property( "process-from-days-ago-default" )
    .description( "References a context property which holds the default number of days ago to start processing from" )
    .required( true )
    .addValidator( NON_EMPTY_VALIDATOR )
    .defaultValue( "#{process-from-days-ago-default}" )
    .build( );

The processor obtains the values of those context parameters, logs them and adds them as as attributes to the outgoing flow file

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

    final String cacheKey = context.getProperty( CACHE_KEY ).evaluateAttributeExpressions( ).getValue( );
    final String daysAgo =  context.getProperty( PROCESS_FROM_DAYS_AGO_DEFAULT ).getValue();

    getLogger()
      .info( String.format( "parameters cache key '%s' days ago %s with attributes %s" ,
        cacheKey ,
        daysAgo ,
        flowFile.getAttributes()
      )
    );

    flowFile = session.putAttribute(flowFile, "added-cache-key", cacheKey );
    flowFile = session.putAttribute(flowFile, "added-days-ago", daysAgo );

    session.transfer( flowFile , SUCCESS);

Note that I use 2 variations on how to access the property, with and without evaluateAttributeExpressions( ). As it is both work, I am unclear if one I should use.

2020-12-10 10:43:18,396 INFO [Timer-Driven Process Thread-3] c.s.n.p.a.adwords.MyParameterProcessor MyParameterProcessor[id=49716825-0176-1000-7c82-90d5d4d51e14] parameters cache key 'ssd-uploader-last-datetime-processed' days ago 30 with attributes {absolute.path=C:\nifi\nifi-1.12.1\data.in/, path=/, filename=ssd-extract.txt, file.lastModifiedTime=2020-12-08T09:42:06+1300, file.creationTime=2020-12-10T10:39:21+1300, file.lastAccessTime=2020-12-10T10:39:21+1300, file.owner=xxxxx, uuid=bc0ec9c5-7f82-40db-888f-ec581aeadb98}

Key: 'added-cache-key' 
Value: 'ssd-uploader-last-datetime-processed'
Key: 'added-days-ago'
Value: '30'

Now how can I test this ( I would have preferred to create the test first but I haven't get tests working !!)? I have this

@Test void testMyParameterProcessor() {

    final MockComponentLog logger = new MockComponentLog( "1234" , this  );

    final TestRunner runner = TestRunners.newTestRunner( MyParameterProcessor.class , logger );

    runner.setValidateExpressionUsage( false );

    // this works
    runner.setProperty( CACHE_KEY.getName() , "ssd-uploader-last-datetime-processed" );
    runner.setProperty( PROCESS_FROM_DAYS_AGO_DEFAULT.getName() , "3" );


    final MockFlowFile flowFile = runner.enqueue( "Hello world!!" .getBytes( ) , Collections.emptyMap( ) );

    runner.run();
    flowFile.assertAttributeEquals( "added-cach-key" , "ssd-uploader-last-datetime-processed" );
    flowFile.assertAttributeEquals( "added-days-ago" , "30" );
    runner.assertAllFlowFilesTransferred( SUCCESS, 1 );
}

Without this line , which I have just discovered ( I didn't notice it in the exception and is undocumented),

runner.setValidateExpressionUsage( false )

the property values cannot be recovered, get this exception

java.lang.AssertionError: java.lang.IllegalStateException: Attempting to Evaluate Expressions but PropertyDescriptor[last-dt-processed-key] 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)

With that line the properties are logged

MyParameterProcessorUTests@487ed8ff parameters cache key 'ssd-uploader-last-datetime-processed' days ago 3 with attributes {path=target, filename=188603694525600.mockFlowFile, uuid=f02f0e04-7901-4460-9fcf-8e1734cd0999}

But the attributes are not added to the flow file. I am using nifi 1.12.1.

Any advice or a pointer to an similar example test will be gratefully received.

Answer 1

Further more I have been trying out the scala-based Nifi integration testing framework plumber which I find very good but it also didn't seem to handle context parameters pl.touk plumber

I revisited this after recognising the runner.setValidateExpressionUsage( false ) answer above. The same can be achieved in plumber by include validateExpressions = false when adding a node: everything works including the attributes being added to the flow file. FYI here is the whole test

class ParameterProcessorITests extends FunSuite with Matchers {

   private val emptyAttributes: Map[String,String] = Map( )

   test("test my parameter processor " ){

val flow = new NifiFlowBuilder()

  .addNode(name = "pp",
    processor = new MyParameterProcessor,
    properties = Map(
      MyParameterProcessor.CACHE_KEY.getName-> "ssd-uploader-last-datetime-processed",
      MyParameterProcessor.PROCESS_FROM_DAYS_AGO_DEFAULT.getName->"3"
    ) ,
    validateExpressions = false
  )

  .addNode("log-message", new LogMessage, emptyAttributes )
  .addNode("log-attribute", new LogAttribute, emptyAttributes )

  .addInputConnection("pp")
  .addConnection("pp", "log-message", MyParameterProcessor.SUCCESS )
  .addConnection("pp", "log-attribute", MyParameterProcessor.SUCCESS )
  .addOutputConnection("pp", MyParameterProcessor.SUCCESS )

  .build()

flow.enqueue(
  "anything".getBytes,
  emptyAttributes
)
flow.run()

val results = flow.executionResult.outputFlowFiles
results.foreach(file => println( new String( file.toByteArray ) ) )
results should have size 1
results.head.assertAttributeEquals("added-cache-key", "ssd-uploader-last-datetime-processed" )
results.head.assertAttributeEquals("added-days-ago", "3" )

} }