http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestQuery.java ---------------------------------------------------------------------- diff --git a/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestQuery.java b/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestQuery.java index 05bd017..9f4bf4d 100644 --- a/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestQuery.java +++ b/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestQuery.java @@ -35,6 +35,7 @@ import java.util.HashSet; import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.Properties; import org.apache.nifi.attribute.expression.language.Query.Range; import org.apache.nifi.attribute.expression.language.evaluation.QueryResult; @@ -43,7 +44,12 @@ import org.apache.nifi.attribute.expression.language.exception.AttributeExpressi import org.apache.nifi.expression.AttributeExpression.ResultType; import org.apache.nifi.flowfile.FlowFile; import org.antlr.runtime.tree.Tree; + +import org.apache.nifi.registry.VariableRegistry; +import org.apache.nifi.registry.VariableRegistryFactory; +import org.apache.nifi.registry.VariableRegistryUtils; import org.junit.Assert; + import org.junit.Ignore; import org.junit.Test; @@ -51,6 +57,7 @@ import org.mockito.Mockito; public class TestQuery { + @Test public void testCompilation() { assertInvalid("${attr:uuid()}"); @@ -114,7 +121,7 @@ public class TestQuery { final Map<String, String> attributes = new HashMap<>(); attributes.put("x", "x"); attributes.put("y", "x"); - final String result = Query.evaluateExpressions(expression, attributes, null); + final String result = Query.evaluateExpressions(expression,VariableRegistryFactory.getInstance(attributes), null); assertEquals("true", result); Query.validateExpression(expression, false); @@ -174,14 +181,14 @@ public class TestQuery { public void testWithTicksOutside() { final Map<String, String> attributes = new HashMap<>(); attributes.put("attr", "My Value"); - + VariableRegistry registry = VariableRegistryFactory.getInstance(attributes); assertEquals(1, Query.extractExpressionRanges("\"${attr}").size()); assertEquals(1, Query.extractExpressionRanges("'${attr}").size()); assertEquals(1, Query.extractExpressionRanges("'${attr}'").size()); assertEquals(1, Query.extractExpressionRanges("${attr}").size()); - assertEquals("'My Value'", Query.evaluateExpressions("'${attr}'", attributes, null)); - assertEquals("'My Value", Query.evaluateExpressions("'${attr}", attributes, null)); + assertEquals("'My Value'", Query.evaluateExpressions("'${attr}'", registry, null)); + assertEquals("'My Value", Query.evaluateExpressions("'${attr}", registry, null)); } @Test @@ -191,7 +198,7 @@ public class TestQuery { final Map<String, String> attributes = new HashMap<>(); attributes.put("dateTime", "2013/11/18 10:22:27.678"); - final QueryResult<?> result = query.evaluate(attributes); + final QueryResult<?> result = query.evaluate(VariableRegistryFactory.getInstance(attributes)); assertEquals(ResultType.NUMBER, result.getResultType()); assertEquals(1384788147678L, result.getValue()); } @@ -220,7 +227,7 @@ public class TestQuery { final Date roundedToNearestSecond = new Date(date.getTime() - millis); final String formatted = sdf.format(roundedToNearestSecond); - final QueryResult<?> result = query.evaluate(attributes); + final QueryResult<?> result = query.evaluate(VariableRegistryFactory.getInstance(attributes)); assertEquals(ResultType.STRING, result.getResultType()); assertEquals(formatted, result.getValue()); } @@ -230,14 +237,31 @@ public class TestQuery { final Map<String, String> attributes = new HashMap<>(); attributes.put("x", "abc"); attributes.put("a", "abc"); + VariableRegistry registry = VariableRegistryFactory.getInstance(attributes); verifyEquals("${x:equals(${a})}", attributes, true); Query.validateExpression("${x:equals('${a}')}", false); - assertEquals("true", Query.evaluateExpressions("${x:equals('${a}')}", attributes, null)); + assertEquals("true", Query.evaluateExpressions("${x:equals('${a}')}", registry, null)); + + Query.validateExpression("${x:equals(\"${a}\")}", false); + assertEquals("true", Query.evaluateExpressions("${x:equals(\"${a}\")}", registry, null)); + } + + @Test + public void testEmbeddedExpressionsAndQuotesWithProperties() { + final Properties attributes = new Properties(); + attributes.put("x", "abc"); + attributes.put("a", "abc"); + VariableRegistry registry = VariableRegistryFactory.getPropertiesInstance(attributes); + + verifyEquals("${x:equals(${a})}",registry,true); + + Query.validateExpression("${x:equals('${a}')}", false); + assertEquals("true", Query.evaluateExpressions("${x:equals('${a}')}", registry, null)); Query.validateExpression("${x:equals(\"${a}\")}", false); - assertEquals("true", Query.evaluateExpressions("${x:equals(\"${a}\")}", attributes, null)); + assertEquals("true", Query.evaluateExpressions("${x:equals(\"${a}\")}", registry, null)); } @Test @@ -343,7 +367,9 @@ public class TestQuery { Mockito.when(mockFlowFile.getSize()).thenReturn(1L); Mockito.when(mockFlowFile.getLineageIdentifiers()).thenReturn(new HashSet<String>()); Mockito.when(mockFlowFile.getLineageStartDate()).thenReturn(System.currentTimeMillis()); - return Query.evaluateExpressions(queryString, mockFlowFile); + + final VariableRegistry variableRegistry = VariableRegistryUtils.populateRegistry(VariableRegistryUtils.createVariableRegistry(),mockFlowFile,null); + return Query.evaluateExpressions(queryString,variableRegistry); } @Test @@ -499,7 +525,7 @@ public class TestQuery { verifyEquals("${x:toNumber():gt( ${y:toNumber():plus( ${z:toNumber()} )} )}", attributes, true); attributes.put("y", "88"); - assertEquals("true", Query.evaluateExpressions("${x:equals( '${y}' )}", attributes, null)); + assertEquals("true", Query.evaluateExpressions("${x:equals( '${y}' )}", VariableRegistryFactory.getInstance(attributes), null)); } @Test @@ -547,7 +573,7 @@ public class TestQuery { final String format = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"; final String query = "startDateTime=\"${date:toNumber():toDate():format(\"" + format + "\")}\""; - final String result = Query.evaluateExpressions(query, attributes, null); + final String result = Query.evaluateExpressions(query, VariableRegistryFactory.getInstance(attributes), null); final String expectedTime = new SimpleDateFormat(format, Locale.US).format(timestamp); assertEquals("startDateTime=\"" + expectedTime + "\"", result); @@ -616,7 +642,7 @@ public class TestQuery { final String query = "${ abc:equals('abc'):or( \n\t${xx:isNull()}\n) }"; assertEquals(ResultType.BOOLEAN, Query.getResultType(query)); Query.validateExpression(query, false); - assertEquals("true", Query.evaluateExpressions(query)); + assertEquals("true", Query.evaluateExpressions(query,VariableRegistryUtils.createVariableRegistry())); } @Test @@ -632,7 +658,7 @@ public class TestQuery { public void testComments() { final Map<String, String> attributes = new HashMap<>(); attributes.put("abc", "xyz"); - + VariableRegistry registry = VariableRegistryFactory.getInstance(attributes); final String expression = "# hello, world\n" + "${# ref attr\n" @@ -643,12 +669,12 @@ public class TestQuery { + "}"; Query query = Query.compile(expression); - QueryResult<?> result = query.evaluate(attributes); + QueryResult<?> result = query.evaluate(registry); assertEquals(ResultType.STRING, result.getResultType()); assertEquals("xyz", result.getValue()); query = Query.compile("${abc:append('# hello') #good-bye \n}"); - result = query.evaluate(attributes); + result = query.evaluate(registry); assertEquals(ResultType.STRING, result.getResultType()); assertEquals("xyz# hello", result.getValue()); } @@ -777,14 +803,15 @@ public class TestQuery { final Map<String, String> attributes = new HashMap<>(); attributes.put("entryDate", String.valueOf(now.getTimeInMillis())); + VariableRegistry registry = VariableRegistryFactory.getInstance(attributes); verifyEquals("${entryDate:toNumber():toDate():format('yyyy')}", attributes, String.valueOf(year)); attributes.clear(); attributes.put("month", "3"); attributes.put("day", "4"); attributes.put("year", "2013"); - assertEquals("63", Query.evaluateExpressions("${year:append('/'):append(${month}):append('/'):append(${day}):toDate('yyyy/MM/dd'):format('D')}", attributes, null)); - assertEquals("63", Query.evaluateExpressions("${year:append('/'):append('${month}'):append('/'):append('${day}'):toDate('yyyy/MM/dd'):format('D')}", attributes, null)); + assertEquals("63", Query.evaluateExpressions("${year:append('/'):append(${month}):append('/'):append(${day}):toDate('yyyy/MM/dd'):format('D')}", registry, null)); + assertEquals("63", Query.evaluateExpressions("${year:append('/'):append('${month}'):append('/'):append('${day}'):toDate('yyyy/MM/dd'):format('D')}", registry, null)); verifyEquals("${year:append('/'):append(${month}):append('/'):append(${day}):toDate('yyyy/MM/dd'):format('D')}", attributes, "63"); } @@ -792,8 +819,9 @@ public class TestQuery { @Test public void testSystemProperty() { System.setProperty("hello", "good-bye"); - assertEquals("good-bye", Query.evaluateExpressions("${hello}")); - assertEquals("good-bye", Query.compile("${hello}").evaluate().getValue()); + VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry(); + assertEquals("good-bye", Query.evaluateExpressions("${hello}",VariableRegistryUtils.createVariableRegistry())); + assertEquals("good-bye", Query.compile("${hello}").evaluate(variableRegistry).getValue()); } @Test @@ -833,14 +861,15 @@ public class TestQuery { final Map<String, String> attributes = new HashMap<>(); attributes.put("abc", "a,b,c"); attributes.put("xyz", "abc"); + VariableRegistry registry = VariableRegistryFactory.getInstance(attributes); final String query = "${anyDelineatedValue('${abc}', ','):equals('b')}"; assertEquals(ResultType.BOOLEAN, Query.getResultType(query)); - assertEquals("true", Query.evaluateExpressions(query, attributes, null)); - assertEquals("true", Query.evaluateExpressions("${anyDelineatedValue('${abc}', ','):equals('a')}", attributes, null)); - assertEquals("true", Query.evaluateExpressions("${anyDelineatedValue('${abc}', ','):equals('c')}", attributes, null)); - assertEquals("false", Query.evaluateExpressions("${anyDelineatedValue('${abc}', ','):equals('d')}", attributes, null)); + assertEquals("true", Query.evaluateExpressions(query, registry, null)); + assertEquals("true", Query.evaluateExpressions("${anyDelineatedValue('${abc}', ','):equals('a')}", registry, null)); + assertEquals("true", Query.evaluateExpressions("${anyDelineatedValue('${abc}', ','):equals('c')}", registry, null)); + assertEquals("false", Query.evaluateExpressions("${anyDelineatedValue('${abc}', ','):equals('d')}", registry, null)); verifyEquals("${anyDelineatedValue(${abc}, ','):equals('b')}", attributes, true); verifyEquals("${anyDelineatedValue(${abc}, ','):equals('a')}", attributes, true); @@ -854,13 +883,15 @@ public class TestQuery { attributes.put("abc", "a,b,c"); attributes.put("xyz", "abc"); + VariableRegistry registry = VariableRegistryFactory.getInstance(attributes); + final String query = "${allDelineatedValues('${abc}', ','):matches('[abc]')}"; assertEquals(ResultType.BOOLEAN, Query.getResultType(query)); - assertEquals("true", Query.evaluateExpressions(query, attributes, null)); - assertEquals("true", Query.evaluateExpressions(query, attributes, null)); - assertEquals("false", Query.evaluateExpressions("${allDelineatedValues('${abc}', ','):matches('[abd]')}", attributes, null)); - assertEquals("false", Query.evaluateExpressions("${allDelineatedValues('${abc}', ','):equals('a'):not()}", attributes, null)); + assertEquals("true", Query.evaluateExpressions(query, registry, null)); + assertEquals("true", Query.evaluateExpressions(query, registry, null)); + assertEquals("false", Query.evaluateExpressions("${allDelineatedValues('${abc}', ','):matches('[abd]')}",registry, null)); + assertEquals("false", Query.evaluateExpressions("${allDelineatedValues('${abc}', ','):equals('a'):not()}", registry, null)); verifyEquals("${allDelineatedValues(${abc}, ','):matches('[abc]')}", attributes, true); verifyEquals("${allDelineatedValues(${abc}, ','):matches('[abd]')}", attributes, false); @@ -926,12 +957,13 @@ public class TestQuery { attributes.put("xyz", "4132"); attributes.put("hello", "world!"); attributes.put("dotted", "abc.xyz"); + VariableRegistry registry = VariableRegistryFactory.getInstance(attributes); - final String evaluated = Query.evaluateExpressions("${abc:matches('1234${end}4321')}", attributes, null); + final String evaluated = Query.evaluateExpressions("${abc:matches('1234${end}4321')}", registry, null); assertEquals("true", evaluated); attributes.put("end", "888"); - final String secondEvaluation = Query.evaluateExpressions("${abc:matches('1234${end}4321')}", attributes, null); + final String secondEvaluation = Query.evaluateExpressions("${abc:matches('1234${end}4321')}", registry, null); assertEquals("false", secondEvaluation); verifyEquals("${dotted:matches('abc\\.xyz')}", attributes, true); @@ -946,11 +978,13 @@ public class TestQuery { attributes.put("hello", "world!"); attributes.put("dotted", "abc.xyz"); - final String evaluated = Query.evaluateExpressions("${abc:find('1234${end}4321')}", attributes, null); + final String evaluated = Query.evaluateExpressions("${abc:find('1234${end}4321')}", VariableRegistryFactory.getInstance(attributes), null); assertEquals("true", evaluated); attributes.put("end", "888"); - final String secondEvaluation = Query.evaluateExpressions("${abc:find('${end}4321')}", attributes, null); + + + final String secondEvaluation = Query.evaluateExpressions("${abc:find('${end}4321')}",VariableRegistryFactory.getInstance(attributes), null); assertEquals("false", secondEvaluation); verifyEquals("${dotted:find('\\.')}", attributes, true); @@ -1131,7 +1165,7 @@ public class TestQuery { attributes.put("b", "x"); attributes.put("abcxcba", "hello"); - final String evaluated = Query.evaluateExpressions("${ 'abc${b}cba':substring(0, 1) }", attributes, null); + final String evaluated = Query.evaluateExpressions("${ 'abc${b}cba':substring(0, 1) }", VariableRegistryFactory.getInstance(attributes), null); assertEquals("h", evaluated); } @@ -1165,7 +1199,7 @@ public class TestQuery { final List<String> expressions = Query.extractExpressions(query); assertEquals(1, expressions.size()); assertEquals("${abc}", expressions.get(0)); - assertEquals("{ xyz }", Query.evaluateExpressions(query, attributes)); + assertEquals("{ xyz }", Query.evaluateExpressions(query, VariableRegistryFactory.getInstance(attributes))); } @Test @@ -1189,7 +1223,7 @@ public class TestQuery { QueryResult<?> getResult(String expr, Map<String, String> attrs) { final Query query = Query.compile(expr); - final QueryResult<?> result = query.evaluate(attrs); + final QueryResult<?> result = query.evaluate(VariableRegistryFactory.getInstance(attrs)); return result; } @@ -1298,11 +1332,17 @@ public class TestQuery { } private void verifyEquals(final String expression, final Map<String, String> attributes, final Object expectedResult) { + + VariableRegistry registry = VariableRegistryFactory.getInstance(attributes); + verifyEquals(expression,registry,expectedResult); + } + + private void verifyEquals(final String expression, final VariableRegistry registry, final Object expectedResult) { Query.validateExpression(expression, false); - assertEquals(String.valueOf(expectedResult), Query.evaluateExpressions(expression, attributes, null)); + assertEquals(String.valueOf(expectedResult), Query.evaluateExpressions(expression, registry, null)); final Query query = Query.compile(expression); - final QueryResult<?> result = query.evaluate(attributes); + final QueryResult<?> result = query.evaluate(registry); if (expectedResult instanceof Number) { assertEquals(ResultType.NUMBER, result.getResultType());
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestStandardPreparedQuery.java ---------------------------------------------------------------------- diff --git a/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestStandardPreparedQuery.java b/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestStandardPreparedQuery.java index 5acba8d..dbee665 100644 --- a/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestStandardPreparedQuery.java +++ b/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestStandardPreparedQuery.java @@ -22,6 +22,7 @@ import java.util.HashMap; import java.util.Map; import java.util.concurrent.TimeUnit; +import org.apache.nifi.registry.VariableRegistryFactory; import org.junit.Ignore; import org.junit.Test; @@ -53,7 +54,7 @@ public class TestStandardPreparedQuery { final StandardPreparedQuery prepared = (StandardPreparedQuery) Query.prepare("${xx}"); final long start = System.nanoTime(); for (int i = 0; i < 10000000; i++) { - assertEquals("world", prepared.evaluateExpressions(attrs, null)); + assertEquals("world", prepared.evaluateExpressions(VariableRegistryFactory.getInstance(attrs), null)); } final long nanos = System.nanoTime() - start; System.out.println(TimeUnit.NANOSECONDS.toMillis(nanos)); @@ -67,7 +68,7 @@ public class TestStandardPreparedQuery { final long start = System.nanoTime(); for (int i = 0; i < 10000000; i++) { - assertEquals("world", Query.evaluateExpressions("${xx}", attrs)); + assertEquals("world", Query.evaluateExpressions("${xx}", VariableRegistryFactory.getInstance(attrs))); } final long nanos = System.nanoTime() - start; System.out.println(TimeUnit.NANOSECONDS.toMillis(nanos)); @@ -85,7 +86,7 @@ public class TestStandardPreparedQuery { } private String evaluate(final String query, final Map<String, String> attrs) { - final String evaluated = ((StandardPreparedQuery) Query.prepare(query)).evaluateExpressions(attrs, null); + final String evaluated = ((StandardPreparedQuery) Query.prepare(query)).evaluateExpressions(VariableRegistryFactory.getInstance(attrs), null); return evaluated; } http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java ---------------------------------------------------------------------- diff --git a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java index 8c98c0b..ecef74b 100644 --- a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java +++ b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java @@ -200,6 +200,9 @@ public class NiFiProperties extends Properties { public static final String STATE_MANAGEMENT_START_EMBEDDED_ZOOKEEPER = "nifi.state.management.embedded.zookeeper.start"; public static final String STATE_MANAGEMENT_ZOOKEEPER_PROPERTIES = "nifi.state.management.embedded.zookeeper.properties"; + // expression language properties + public static final String VARIABLE_REGISTRY_PROPERTIES = "nifi.variable.registry.properties"; + // defaults public static final String DEFAULT_TITLE = "NiFi"; public static final Boolean DEFAULT_AUTO_RESUME_STATE = true; @@ -248,11 +251,13 @@ public class NiFiProperties extends Properties { public static final String DEFAULT_CLUSTER_MANAGER_SAFEMODE_DURATION = "0 sec"; // state management defaults + public static final String DEFAULT_STATE_MANAGEMENT_CONFIG_FILE = "conf/state-management.xml"; // Kerberos defaults public static final String DEFAULT_KERBEROS_AUTHENTICATION_EXPIRATION = "12 hours"; + private NiFiProperties() { super(); } @@ -1073,4 +1078,28 @@ public class NiFiProperties extends Properties { public boolean isStartEmbeddedZooKeeper() { return Boolean.parseBoolean(getProperty(STATE_MANAGEMENT_START_EMBEDDED_ZOOKEEPER)); } + + public String getVariableRegistryProperties(){ + return getProperty(VARIABLE_REGISTRY_PROPERTIES); + } + + public Path[] getVariableRegistryPropertiesPaths() { + final List<Path> vrPropertiesPaths = new ArrayList<>(); + + final String vrPropertiesFiles = getVariableRegistryProperties(); + if(!StringUtils.isEmpty(vrPropertiesFiles)) { + + final List<String> vrPropertiesFileList = Arrays.asList(vrPropertiesFiles.split(",")); + + for(String propertiesFile : vrPropertiesFileList){ + vrPropertiesPaths.add(Paths.get(propertiesFile)); + } + + return vrPropertiesPaths.toArray( new Path[vrPropertiesPaths.size()]); + } else { + return null; + } + } + + } http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/MockConfigurationContext.java ---------------------------------------------------------------------- diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockConfigurationContext.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockConfigurationContext.java index c90e722..4c3b399 100644 --- a/nifi-mock/src/main/java/org/apache/nifi/util/MockConfigurationContext.java +++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockConfigurationContext.java @@ -25,21 +25,26 @@ import org.apache.nifi.components.PropertyValue; import org.apache.nifi.controller.ConfigurationContext; import org.apache.nifi.controller.ControllerService; import org.apache.nifi.controller.ControllerServiceLookup; +import org.apache.nifi.registry.VariableRegistry; public class MockConfigurationContext implements ConfigurationContext { private final Map<PropertyDescriptor, String> properties; private final ControllerServiceLookup serviceLookup; private final ControllerService service; + private final VariableRegistry variableRegistry; - public MockConfigurationContext(final Map<PropertyDescriptor, String> properties, final ControllerServiceLookup serviceLookup) { - this(null, properties, serviceLookup); + public MockConfigurationContext(final Map<PropertyDescriptor, String> properties, final ControllerServiceLookup serviceLookup, + final VariableRegistry variableRegistry) { + this(null, properties, serviceLookup, variableRegistry); } - public MockConfigurationContext(final ControllerService service, final Map<PropertyDescriptor, String> properties, final ControllerServiceLookup serviceLookup) { + public MockConfigurationContext(final ControllerService service, final Map<PropertyDescriptor, String> properties, final ControllerServiceLookup serviceLookup, + final VariableRegistry variableRegistry) { this.service = service; this.properties = properties; this.serviceLookup = serviceLookup; + this.variableRegistry = variableRegistry; } @Override @@ -48,7 +53,7 @@ public class MockConfigurationContext implements ConfigurationContext { if (value == null) { value = getActualDescriptor(property).getDefaultValue(); } - return new MockPropertyValue(value, serviceLookup); + return new MockPropertyValue(value, serviceLookup, variableRegistry); } @Override http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceInitializationContext.java ---------------------------------------------------------------------- diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceInitializationContext.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceInitializationContext.java index 754bec0..3134c19 100644 --- a/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceInitializationContext.java +++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceInitializationContext.java @@ -68,4 +68,6 @@ public class MockControllerServiceInitializationContext extends MockControllerSe public StateManager getStateManager() { return stateManager; } + + } http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceLookup.java ---------------------------------------------------------------------- diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceLookup.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceLookup.java index 219ee24..fdb40a8 100644 --- a/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceLookup.java +++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceLookup.java @@ -97,4 +97,5 @@ public abstract class MockControllerServiceLookup implements ControllerServiceLo final ControllerServiceConfiguration status = controllerServiceMap.get(serviceIdentifier); return status == null ? null : serviceIdentifier; } + } http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessContext.java ---------------------------------------------------------------------- diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessContext.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessContext.java index d50d708..c88de5f 100644 --- a/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessContext.java +++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessContext.java @@ -40,6 +40,7 @@ import org.apache.nifi.controller.ControllerServiceLookup; import org.apache.nifi.processor.Processor; import org.apache.nifi.processor.Relationship; import org.apache.nifi.processor.SchedulingContext; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.state.MockStateManager; import org.junit.Assert; @@ -48,6 +49,7 @@ public class MockProcessContext extends MockControllerServiceLookup implements S private final ConfigurableComponent component; private final Map<PropertyDescriptor, String> properties = new HashMap<>(); private final StateManager stateManager; + private final VariableRegistry variableRegistry; private String annotationData = null; private boolean yieldCalled = false; @@ -60,22 +62,24 @@ public class MockProcessContext extends MockControllerServiceLookup implements S private volatile Set<Relationship> connections = new HashSet<>(); private volatile Set<Relationship> unavailableRelationships = new HashSet<>(); - public MockProcessContext(final ConfigurableComponent component) { - this(component, new MockStateManager(component)); + public MockProcessContext(final ConfigurableComponent component, final VariableRegistry variableRegistry) { + this(component, new MockStateManager(component), variableRegistry); } /** * Creates a new MockProcessContext for the given Processor * * @param component being mocked + * @param variableRegistry variableRegistry */ - public MockProcessContext(final ConfigurableComponent component, final StateManager stateManager) { + public MockProcessContext(final ConfigurableComponent component, final StateManager stateManager, final VariableRegistry variableRegistry) { this.component = Objects.requireNonNull(component); this.stateManager = stateManager; + this.variableRegistry = variableRegistry; } - public MockProcessContext(final ControllerService component, final MockProcessContext context, final StateManager stateManager) { - this(component, stateManager); + public MockProcessContext(final ControllerService component, final MockProcessContext context, final StateManager stateManager, final VariableRegistry variableRegistry) { + this(component, stateManager, variableRegistry); try { annotationData = context.getControllerServiceAnnotationData(component); @@ -102,12 +106,13 @@ public class MockProcessContext extends MockControllerServiceLookup implements S final String setPropertyValue = properties.get(descriptor); final String propValue = (setPropertyValue == null) ? descriptor.getDefaultValue() : setPropertyValue; - return new MockPropertyValue(propValue, this, (enableExpressionValidation && allowExpressionValidation) ? descriptor : null); + + return new MockPropertyValue(propValue, this, variableRegistry, (enableExpressionValidation && allowExpressionValidation) ? descriptor : null); } @Override public PropertyValue newPropertyValue(final String rawValue) { - return new MockPropertyValue(rawValue, this); + return new MockPropertyValue(rawValue, this, variableRegistry); } public ValidationResult setProperty(final String propertyName, final String propertyValue) { @@ -130,7 +135,7 @@ public class MockProcessContext extends MockControllerServiceLookup implements S requireNonNull(value, "Cannot set property to null value; if the intent is to remove the property, call removeProperty instead"); final PropertyDescriptor fullyPopulatedDescriptor = component.getPropertyDescriptor(descriptor.getName()); - final ValidationResult result = fullyPopulatedDescriptor.validate(value, new MockValidationContext(this, stateManager)); + final ValidationResult result = fullyPopulatedDescriptor.validate(value, new MockValidationContext(this, stateManager, variableRegistry)); String oldValue = properties.put(fullyPopulatedDescriptor, value); if (oldValue == null) { oldValue = fullyPopulatedDescriptor.getDefaultValue(); @@ -213,7 +218,7 @@ public class MockProcessContext extends MockControllerServiceLookup implements S * non-null */ public Collection<ValidationResult> validate() { - return component.validate(new MockValidationContext(this, stateManager)); + return component.validate(new MockValidationContext(this, stateManager, variableRegistry)); } public boolean isValid() { http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java ---------------------------------------------------------------------- diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java index 6e94943..9ef513c 100644 --- a/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java +++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java @@ -80,4 +80,5 @@ public class MockProcessorInitializationContext implements ProcessorInitializati public boolean isControllerServiceEnabling(String serviceIdentifier) { return context.isControllerServiceEnabling(serviceIdentifier); } + } http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/MockPropertyValue.java ---------------------------------------------------------------------- diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockPropertyValue.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockPropertyValue.java index 090a8eb..0fb4f89 100644 --- a/nifi-mock/src/main/java/org/apache/nifi/util/MockPropertyValue.java +++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockPropertyValue.java @@ -28,6 +28,7 @@ import org.apache.nifi.expression.AttributeValueDecorator; import org.apache.nifi.flowfile.FlowFile; import org.apache.nifi.processor.DataUnit; import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.registry.VariableRegistry; public class MockPropertyValue implements PropertyValue { private final String rawValue; @@ -35,24 +36,27 @@ public class MockPropertyValue implements PropertyValue { private final ControllerServiceLookup serviceLookup; private final PropertyDescriptor propertyDescriptor; private final PropertyValue stdPropValue; + private final VariableRegistry variableRegistry; private boolean expressionsEvaluated = false; - public MockPropertyValue(final String rawValue, final ControllerServiceLookup serviceLookup) { - this(rawValue, serviceLookup, null); + public MockPropertyValue(final String rawValue, final ControllerServiceLookup serviceLookup, final VariableRegistry variableRegistry) { + this(rawValue, serviceLookup, variableRegistry, null); } - public MockPropertyValue(final String rawValue, final ControllerServiceLookup serviceLookup, final PropertyDescriptor propertyDescriptor) { - this(rawValue, serviceLookup, propertyDescriptor, false); + public MockPropertyValue(final String rawValue, final ControllerServiceLookup serviceLookup, VariableRegistry variableRegistry, final PropertyDescriptor propertyDescriptor) { + this(rawValue, serviceLookup, propertyDescriptor, false, variableRegistry); } - private MockPropertyValue(final String rawValue, final ControllerServiceLookup serviceLookup, final PropertyDescriptor propertyDescriptor, final boolean alreadyEvaluated) { - this.stdPropValue = new StandardPropertyValue(rawValue, serviceLookup); + private MockPropertyValue(final String rawValue, final ControllerServiceLookup serviceLookup, final PropertyDescriptor propertyDescriptor, final boolean alreadyEvaluated, + final VariableRegistry variableRegistry) { + this.stdPropValue = new StandardPropertyValue(rawValue, serviceLookup, variableRegistry); this.rawValue = rawValue; this.serviceLookup = serviceLookup; this.expectExpressions = propertyDescriptor == null ? null : propertyDescriptor.isExpressionLanguageSupported(); this.propertyDescriptor = propertyDescriptor; this.expressionsEvaluated = alreadyEvaluated; + this.variableRegistry = variableRegistry; } @@ -165,7 +169,7 @@ public class MockPropertyValue implements PropertyValue { } final PropertyValue newValue = stdPropValue.evaluateAttributeExpressions(flowFile, additionalAttributes, decorator); - return new MockPropertyValue(newValue.getValue(), serviceLookup, propertyDescriptor, true); + return new MockPropertyValue(newValue.getValue(), serviceLookup, propertyDescriptor, true, variableRegistry); } @Override http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/MockReportingContext.java ---------------------------------------------------------------------- diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockReportingContext.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockReportingContext.java index 33719ec..da43c62 100644 --- a/nifi-mock/src/main/java/org/apache/nifi/util/MockReportingContext.java +++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockReportingContext.java @@ -27,6 +27,7 @@ import org.apache.nifi.components.PropertyValue; import org.apache.nifi.components.state.StateManager; import org.apache.nifi.controller.ControllerService; import org.apache.nifi.controller.ControllerServiceLookup; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.reporting.Bulletin; import org.apache.nifi.reporting.BulletinFactory; import org.apache.nifi.reporting.BulletinRepository; @@ -39,12 +40,14 @@ public class MockReportingContext extends MockControllerServiceLookup implements private final MockEventAccess eventAccess = new MockEventAccess(); private final Map<PropertyDescriptor, String> properties = new HashMap<>(); private final StateManager stateManager; + private final VariableRegistry variableRegistry; private final Map<String, List<Bulletin>> componentBulletinsCreated = new HashMap<>(); - public MockReportingContext(final Map<String, ControllerService> controllerServices, final StateManager stateManager) { + public MockReportingContext(final Map<String, ControllerService> controllerServices, final StateManager stateManager, final VariableRegistry variableRegistry) { this.controllerServices = new HashMap<>(); this.stateManager = stateManager; + this.variableRegistry = variableRegistry; for (final Map.Entry<String, ControllerService> entry : controllerServices.entrySet()) { this.controllerServices.put(entry.getKey(), new ControllerServiceConfiguration(entry.getValue())); } @@ -58,7 +61,7 @@ public class MockReportingContext extends MockControllerServiceLookup implements @Override public PropertyValue getProperty(final PropertyDescriptor property) { final String configuredValue = properties.get(property); - return new MockPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, this); + return new MockPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, this, variableRegistry); } public void setProperty(final String propertyName, final String value) { http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java ---------------------------------------------------------------------- diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java index 6442778..322288b 100644 --- a/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java +++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java @@ -31,16 +31,20 @@ import org.apache.nifi.components.state.StateManager; import org.apache.nifi.controller.ControllerService; import org.apache.nifi.controller.ControllerServiceLookup; import org.apache.nifi.expression.ExpressionLanguageCompiler; +import org.apache.nifi.registry.VariableRegistry; + public class MockValidationContext implements ValidationContext, ControllerServiceLookup { private final MockProcessContext context; private final Map<String, Boolean> expressionLanguageSupported; private final StateManager stateManager; + private final VariableRegistry variableRegistry; - public MockValidationContext(final MockProcessContext processContext, final StateManager stateManager) { + public MockValidationContext(final MockProcessContext processContext, final StateManager stateManager, final VariableRegistry variableRegistry) { this.context = processContext; this.stateManager = stateManager; + this.variableRegistry = variableRegistry; final Map<PropertyDescriptor, String> properties = processContext.getProperties(); expressionLanguageSupported = new HashMap<>(properties.size()); @@ -56,18 +60,18 @@ public class MockValidationContext implements ValidationContext, ControllerServi @Override public PropertyValue newPropertyValue(final String rawValue) { - return new MockPropertyValue(rawValue, this); + return new MockPropertyValue(rawValue, this, variableRegistry); } @Override public ExpressionLanguageCompiler newExpressionLanguageCompiler() { - return new StandardExpressionLanguageCompiler(); + return new StandardExpressionLanguageCompiler(variableRegistry); } @Override public ValidationContext getControllerServiceValidationContext(final ControllerService controllerService) { - final MockProcessContext serviceProcessContext = new MockProcessContext(controllerService, context, stateManager); - return new MockValidationContext(serviceProcessContext, stateManager); + final MockProcessContext serviceProcessContext = new MockProcessContext(controllerService, context, stateManager, variableRegistry); + return new MockValidationContext(serviceProcessContext, stateManager, variableRegistry); } @Override @@ -136,4 +140,5 @@ public class MockValidationContext implements ValidationContext, ControllerServi final Boolean supported = expressionLanguageSupported.get(propertyName); return Boolean.TRUE.equals(supported); } + } http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java ---------------------------------------------------------------------- diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java b/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java index 43621e0..6206c02 100644 --- a/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java +++ b/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java @@ -69,6 +69,7 @@ import org.apache.nifi.processor.Processor; import org.apache.nifi.processor.Relationship; import org.apache.nifi.provenance.ProvenanceEventRecord; import org.apache.nifi.provenance.ProvenanceReporter; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.reporting.InitializationException; import org.apache.nifi.state.MockStateManager; import org.junit.Assert; @@ -84,6 +85,7 @@ public class StandardProcessorTestRunner implements TestRunner { private final boolean triggerSerially; private final MockStateManager processorStateManager; private final Map<String, MockStateManager> controllerServiceStateManagers = new HashMap<>(); + private final VariableRegistry variableRegistry; private int numThreads = 1; private final AtomicInteger invocations = new AtomicInteger(0); @@ -100,14 +102,15 @@ public class StandardProcessorTestRunner implements TestRunner { populateDeprecatedMethods(); } - StandardProcessorTestRunner(final Processor processor) { + StandardProcessorTestRunner(final Processor processor, final VariableRegistry variableRegistry) { this.processor = processor; this.idGenerator = new AtomicLong(0L); this.sharedState = new SharedSessionState(processor, idGenerator); this.flowFileQueue = sharedState.getFlowFileQueue(); this.sessionFactory = new MockSessionFactory(sharedState, processor); this.processorStateManager = new MockStateManager(processor); - this.context = new MockProcessContext(processor, processorStateManager); + this.variableRegistry = variableRegistry; + this.context = new MockProcessContext(processor, processorStateManager, variableRegistry); detectDeprecatedAnnotations(processor); @@ -670,7 +673,7 @@ public class StandardProcessorTestRunner implements TestRunner { throw new IllegalStateException("Controller Service has not been added to this TestRunner via the #addControllerService method"); } - final ValidationContext validationContext = new MockValidationContext(context, serviceStateManager).getControllerServiceValidationContext(service); + final ValidationContext validationContext = new MockValidationContext(context, serviceStateManager, variableRegistry).getControllerServiceValidationContext(service); final Collection<ValidationResult> results = context.getControllerService(service.getIdentifier()).validate(validationContext); for (final ValidationResult result : results) { @@ -689,7 +692,7 @@ public class StandardProcessorTestRunner implements TestRunner { throw new IllegalStateException("Controller Service has not been added to this TestRunner via the #addControllerService method"); } - final ValidationContext validationContext = new MockValidationContext(context, serviceStateManager).getControllerServiceValidationContext(service); + final ValidationContext validationContext = new MockValidationContext(context, serviceStateManager, variableRegistry).getControllerServiceValidationContext(service); final Collection<ValidationResult> results = context.getControllerService(service.getIdentifier()).validate(validationContext); for (final ValidationResult result : results) { @@ -732,7 +735,7 @@ public class StandardProcessorTestRunner implements TestRunner { } try { - final ConfigurationContext configContext = new MockConfigurationContext(service, configuration.getProperties(), context); + final ConfigurationContext configContext = new MockConfigurationContext(service, configuration.getProperties(), context,variableRegistry); ReflectionUtils.invokeMethodsWithAnnotation(OnEnabled.class, service, configContext); } catch (final InvocationTargetException ite) { ite.getCause().printStackTrace(); @@ -804,7 +807,7 @@ public class StandardProcessorTestRunner implements TestRunner { final Map<PropertyDescriptor, String> curProps = configuration.getProperties(); final Map<PropertyDescriptor, String> updatedProps = new HashMap<>(curProps); - final ValidationContext validationContext = new MockValidationContext(context, serviceStateManager).getControllerServiceValidationContext(service); + final ValidationContext validationContext = new MockValidationContext(context, serviceStateManager, variableRegistry).getControllerServiceValidationContext(service); final ValidationResult validationResult = property.validate(value, validationContext); updatedProps.put(property, value); http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java ---------------------------------------------------------------------- diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java b/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java index f2b0b23..f7153bf 100644 --- a/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java +++ b/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java @@ -17,11 +17,12 @@ package org.apache.nifi.util; import org.apache.nifi.processor.Processor; +import org.apache.nifi.registry.VariableRegistryUtils; public class TestRunners { public static TestRunner newTestRunner(final Processor processor) { - return new StandardProcessorTestRunner(processor); + return new StandardProcessorTestRunner(processor, VariableRegistryUtils.createVariableRegistry()); } public static TestRunner newTestRunner(final Class<? extends Processor> processorClass) { http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/test/java/org/apache/nifi/util/CurrentTestStandardProcessorTestRunner.java ---------------------------------------------------------------------- diff --git a/nifi-mock/src/test/java/org/apache/nifi/util/CurrentTestStandardProcessorTestRunner.java b/nifi-mock/src/test/java/org/apache/nifi/util/CurrentTestStandardProcessorTestRunner.java index 6b403af..24e9307 100644 --- a/nifi-mock/src/test/java/org/apache/nifi/util/CurrentTestStandardProcessorTestRunner.java +++ b/nifi-mock/src/test/java/org/apache/nifi/util/CurrentTestStandardProcessorTestRunner.java @@ -20,6 +20,7 @@ import org.apache.nifi.processor.AbstractProcessor; import org.apache.nifi.processor.ProcessContext; import org.apache.nifi.processor.ProcessSession; import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.registry.VariableRegistryUtils; import org.junit.Assert; import org.junit.Test; @@ -31,7 +32,7 @@ public class CurrentTestStandardProcessorTestRunner { @Test public void testOnScheduledCalledAfterRunFinished() { SlowRunProcessor processor = new SlowRunProcessor(); - StandardProcessorTestRunner runner = new StandardProcessorTestRunner(processor); + StandardProcessorTestRunner runner = new StandardProcessorTestRunner(processor, VariableRegistryUtils.createVariableRegistry()); final int iterations = 5; runner.run(iterations); // if the counter is not equal to iterations, the the processor must have been unscheduled http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/test/java/org/apache/nifi/util/TestMockProcessContext.java ---------------------------------------------------------------------- diff --git a/nifi-mock/src/test/java/org/apache/nifi/util/TestMockProcessContext.java b/nifi-mock/src/test/java/org/apache/nifi/util/TestMockProcessContext.java index d48af63..e033451 100644 --- a/nifi-mock/src/test/java/org/apache/nifi/util/TestMockProcessContext.java +++ b/nifi-mock/src/test/java/org/apache/nifi/util/TestMockProcessContext.java @@ -32,6 +32,7 @@ import org.apache.nifi.processor.ProcessContext; import org.apache.nifi.processor.ProcessSession; import org.apache.nifi.processor.exception.ProcessException; import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.registry.VariableRegistryUtils; import org.junit.Test; public class TestMockProcessContext { @@ -39,7 +40,7 @@ public class TestMockProcessContext { @Test public void testRemoveProperty() { final DummyProcessor proc = new DummyProcessor(); - final MockProcessContext context = new MockProcessContext(proc); + final MockProcessContext context = new MockProcessContext(proc, VariableRegistryUtils.createVariableRegistry()); context.setProperty(DummyProcessor.REQUIRED_PROP, "req-value"); context.setProperty(DummyProcessor.OPTIONAL_PROP, "opt-value"); context.setProperty(DummyProcessor.DEFAULTED_PROP, "custom-value"); http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java ---------------------------------------------------------------------- diff --git a/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java b/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java index c2ab654..89d2006 100644 --- a/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java +++ b/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java @@ -30,6 +30,7 @@ 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.registry.VariableRegistryUtils; import org.junit.Ignore; import org.junit.Test; @@ -106,25 +107,25 @@ public class TestStandardProcessorTestRunner { @Test(expected = AssertionError.class) @Ignore("This should not be enabled until we actually fail processor unit tests for using deprecated methods") public void testFailOnDeprecatedTypeAnnotation() { - new StandardProcessorTestRunner(new DeprecatedAnnotation()); + new StandardProcessorTestRunner(new DeprecatedAnnotation(), VariableRegistryUtils.createVariableRegistry()); } @Test @Ignore("This should not be enabled until we actually fail processor unit tests for using deprecated methods") public void testDoesNotFailOnNonDeprecatedTypeAnnotation() { - new StandardProcessorTestRunner(new NewAnnotation()); + new StandardProcessorTestRunner(new NewAnnotation(), VariableRegistryUtils.createVariableRegistry()); } @Test(expected = AssertionError.class) @Ignore("This should not be enabled until we actually fail processor unit tests for using deprecated methods") public void testFailOnDeprecatedMethodAnnotation() { - new StandardProcessorTestRunner(new DeprecatedMethodAnnotation()); + new StandardProcessorTestRunner(new DeprecatedMethodAnnotation(), VariableRegistryUtils.createVariableRegistry()); } @Test @Ignore("This should not be enabled until we actually fail processor unit tests for using deprecated methods") public void testDoesNotFailOnNonDeprecatedMethodAnnotation() { - new StandardProcessorTestRunner(new NewMethodAnnotation()); + new StandardProcessorTestRunner(new NewMethodAnnotation(), VariableRegistryUtils.createVariableRegistry()); } @SuppressWarnings("deprecation") http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-ambari-bundle/nifi-ambari-reporting-task/src/test/java/org/apache/nifi/reporting/ambari/TestAmbariReportingTask.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-ambari-bundle/nifi-ambari-reporting-task/src/test/java/org/apache/nifi/reporting/ambari/TestAmbariReportingTask.java b/nifi-nar-bundles/nifi-ambari-bundle/nifi-ambari-reporting-task/src/test/java/org/apache/nifi/reporting/ambari/TestAmbariReportingTask.java index ce5f8a6..3688b7c 100644 --- a/nifi-nar-bundles/nifi-ambari-bundle/nifi-ambari-reporting-task/src/test/java/org/apache/nifi/reporting/ambari/TestAmbariReportingTask.java +++ b/nifi-nar-bundles/nifi-ambari-bundle/nifi-ambari-reporting-task/src/test/java/org/apache/nifi/reporting/ambari/TestAmbariReportingTask.java @@ -17,13 +17,17 @@ package org.apache.nifi.reporting.ambari; import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.controller.ControllerServiceLookup; import org.apache.nifi.controller.status.ProcessGroupStatus; import org.apache.nifi.controller.status.ProcessorStatus; import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.registry.VariableRegistry; +import org.apache.nifi.registry.VariableRegistryUtils; import org.apache.nifi.reporting.EventAccess; import org.apache.nifi.reporting.InitializationException; import org.apache.nifi.reporting.ReportingContext; import org.apache.nifi.reporting.ReportingInitializationContext; +import org.apache.nifi.util.MockControllerServiceLookup; import org.apache.nifi.util.MockPropertyValue; import org.junit.Before; import org.junit.Test; @@ -43,6 +47,7 @@ import java.util.UUID; public class TestAmbariReportingTask { private ProcessGroupStatus status; + private VariableRegistry variableRegistry; @Before public void setup() { @@ -73,6 +78,7 @@ public class TestAmbariReportingTask { Collection<ProcessGroupStatus> groupStatuses = new ArrayList<>(); groupStatuses.add(groupStatus); status.setProcessGroupStatus(groupStatuses); + variableRegistry = VariableRegistryUtils.createVariableRegistry(); } @Test @@ -101,15 +107,17 @@ public class TestAmbariReportingTask { // mock the ConfigurationContext for setup(...) final ConfigurationContext configurationContext = Mockito.mock(ConfigurationContext.class); + final ControllerServiceLookup serviceLookup = new MockControllerServiceLookup() {}; // mock the ReportingContext for onTrigger(...) final ReportingContext context = Mockito.mock(ReportingContext.class); Mockito.when(context.getProperty(AmbariReportingTask.METRICS_COLLECTOR_URL)) - .thenReturn(new MockPropertyValue(metricsUrl, null)); + .thenReturn(new MockPropertyValue(metricsUrl, serviceLookup, variableRegistry)); Mockito.when(context.getProperty(AmbariReportingTask.APPLICATION_ID)) - .thenReturn(new MockPropertyValue(applicationId, null)); + .thenReturn(new MockPropertyValue(applicationId, serviceLookup, variableRegistry)); Mockito.when(context.getProperty(AmbariReportingTask.HOSTNAME)) - .thenReturn(new MockPropertyValue(hostName, null)); + .thenReturn(new MockPropertyValue(hostName, serviceLookup, variableRegistry)); + final EventAccess eventAccess = Mockito.mock(EventAccess.class); Mockito.when(context.getEventAccess()).thenReturn(eventAccess); @@ -121,7 +129,6 @@ public class TestAmbariReportingTask { task.setup(configurationContext); task.onTrigger(context); } - // override the creation of the client to provide a mock private class TestableAmbariReportingTask extends AmbariReportingTask { http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestFetchElasticsearch.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestFetchElasticsearch.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestFetchElasticsearch.java index cb928fa..4bc8d2c 100644 --- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestFetchElasticsearch.java +++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestFetchElasticsearch.java @@ -18,6 +18,8 @@ package org.apache.nifi.processors.elasticsearch; import org.apache.nifi.processor.ProcessContext; import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.registry.VariableRegistry; +import org.apache.nifi.registry.VariableRegistryUtils; import org.apache.nifi.ssl.SSLContextService; import org.apache.nifi.util.MockFlowFile; import org.apache.nifi.util.MockProcessContext; @@ -62,11 +64,13 @@ public class TestFetchElasticsearch { private InputStream docExample; private TestRunner runner; + private VariableRegistry variableRegistry; @Before public void setUp() throws IOException { ClassLoader classloader = Thread.currentThread().getContextClassLoader(); docExample = classloader.getResourceAsStream("DocumentExample.json"); + variableRegistry = VariableRegistryUtils.createVariableRegistry(); } @@ -216,7 +220,7 @@ public class TestFetchElasticsearch { } }; - MockProcessContext context = new MockProcessContext(processor); + MockProcessContext context = new MockProcessContext(processor, variableRegistry); processor.initialize(new MockProcessorInitializationContext(processor, context)); processor.callCreateElasticsearchClient(context); } http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredReportingContext.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredReportingContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredReportingContext.java index 7f176b0..1ff74d5 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredReportingContext.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredReportingContext.java @@ -34,6 +34,7 @@ import org.apache.nifi.controller.status.ProcessGroupStatus; import org.apache.nifi.controller.status.ProcessorStatus; import org.apache.nifi.controller.status.RemoteProcessGroupStatus; import org.apache.nifi.events.BulletinFactory; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.reporting.Bulletin; import org.apache.nifi.reporting.BulletinRepository; import org.apache.nifi.reporting.ComponentType; @@ -49,14 +50,16 @@ public class ClusteredReportingContext implements ReportingContext { private final Map<PropertyDescriptor, String> properties; private final Map<PropertyDescriptor, PreparedQuery> preparedQueries; private final StateManager stateManager; + private final VariableRegistry variableRegistry; public ClusteredReportingContext(final EventAccess eventAccess, final BulletinRepository bulletinRepository, final Map<PropertyDescriptor, String> properties, - final ControllerServiceProvider serviceProvider, final StateManager stateManager) { + final ControllerServiceProvider serviceProvider, final StateManager stateManager, final VariableRegistry variableRegistry) { this.eventAccess = eventAccess; this.bulletinRepository = bulletinRepository; this.properties = Collections.unmodifiableMap(properties); this.serviceProvider = serviceProvider; this.stateManager = stateManager; + this.variableRegistry = variableRegistry; preparedQueries = new HashMap<>(); for (final Map.Entry<PropertyDescriptor, String> entry : properties.entrySet()) { @@ -104,7 +107,7 @@ public class ClusteredReportingContext implements ReportingContext { @Override public PropertyValue getProperty(final PropertyDescriptor property) { final String configuredValue = properties.get(property); - return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, serviceProvider, preparedQueries.get(property)); + return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, serviceProvider, preparedQueries.get(property), variableRegistry); } @Override http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java index 7bf8de3..b74ea50 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java @@ -181,6 +181,9 @@ import org.apache.nifi.nar.NarCloseable; import org.apache.nifi.nar.NarThreadContextClassLoader; import org.apache.nifi.processor.SimpleProcessLogger; import org.apache.nifi.processor.StandardValidationContextFactory; +import org.apache.nifi.registry.VariableRegistry; +import org.apache.nifi.registry.VariableRegistryFactory; +import org.apache.nifi.registry.VariableRegistryUtils; import org.apache.nifi.remote.RemoteResourceManager; import org.apache.nifi.remote.RemoteSiteListener; import org.apache.nifi.remote.SocketRemoteSiteListener; @@ -382,6 +385,7 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C private final StandardProcessScheduler processScheduler; private final StateManagerProvider stateManagerProvider; private final long componentStatusSnapshotMillis; + private final VariableRegistry variableRegistry; public WebClusterManager(final HttpRequestReplicator httpRequestReplicator, final HttpResponseMapper httpResponseMapper, @@ -412,6 +416,7 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C this.optimisticLockingManager = optimisticLockingManager; senderListener.addHandler(this); senderListener.setBulletinRepository(bulletinRepository); + this.variableRegistry = createVariableRegistry(properties); final String snapshotFrequency = properties.getProperty(NiFiProperties.COMPONENT_STATUS_SNAPSHOT_FREQUENCY, NiFiProperties.DEFAULT_COMPONENT_STATUS_SNAPSHOT_FREQUENCY); long snapshotMillis; @@ -446,7 +451,7 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C reportingTaskEngine = new FlowEngine(8, "Reporting Task Thread"); try { - this.stateManagerProvider = StandardStateManagerProvider.create(properties); + this.stateManagerProvider = StandardStateManagerProvider.create(properties,variableRegistry); } catch (final IOException e) { throw new RuntimeException(e); } @@ -455,17 +460,19 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C @Override public void heartbeat() { } - }, this, encryptor, stateManagerProvider); + }, this, encryptor, stateManagerProvider, variableRegistry); // When we construct the scheduling agents, we can pass null for a lot of the arguments because we are only // going to be scheduling Reporting Tasks. Otherwise, it would not be okay. - processScheduler.setSchedulingAgent(SchedulingStrategy.TIMER_DRIVEN, new TimerDrivenSchedulingAgent(null, reportingTaskEngine, null, encryptor)); - processScheduler.setSchedulingAgent(SchedulingStrategy.CRON_DRIVEN, new QuartzSchedulingAgent(null, reportingTaskEngine, null, encryptor)); + processScheduler.setSchedulingAgent(SchedulingStrategy.TIMER_DRIVEN, new TimerDrivenSchedulingAgent(null, reportingTaskEngine, null, encryptor, variableRegistry)); + processScheduler.setSchedulingAgent(SchedulingStrategy.CRON_DRIVEN, new QuartzSchedulingAgent(null, reportingTaskEngine, null, encryptor, variableRegistry)); processScheduler.setMaxThreadCount(SchedulingStrategy.TIMER_DRIVEN, 10); processScheduler.setMaxThreadCount(SchedulingStrategy.CRON_DRIVEN, 10); processScheduler.scheduleFrameworkTask(new CaptureComponentMetrics(), "Capture Component Metrics", componentStatusSnapshotMillis, componentStatusSnapshotMillis, TimeUnit.MILLISECONDS); - controllerServiceProvider = new StandardControllerServiceProvider(processScheduler, bulletinRepository, stateManagerProvider); + controllerServiceProvider = new StandardControllerServiceProvider(processScheduler, bulletinRepository, stateManagerProvider, variableRegistry); + + } public void start() throws IOException { @@ -956,6 +963,17 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C reconnectionThread.start(); } + private VariableRegistry createVariableRegistry(NiFiProperties properties){ + VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry(); + try { + VariableRegistry customRegistry = VariableRegistryFactory.getPropertiesInstance(properties.getVariableRegistryPropertiesPaths()); + variableRegistry.addRegistry(customRegistry); + } catch (IOException ioe){ + logger.error("Exception thrown while attempting to add properties to registry",ioe); + } + return variableRegistry; + } + private Map<String, ReportingTaskNode> loadReportingTasks(final byte[] serialized) { final Map<String, ReportingTaskNode> tasks = new HashMap<>(); @@ -1105,10 +1123,10 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C } } - final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(this); + final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(this, variableRegistry); final ReportingTaskNode taskNode = new ClusteredReportingTaskNode(task, id, processScheduler, new ClusteredEventAccess(this, auditService), bulletinRepository, controllerServiceProvider, - validationContextFactory, stateManagerProvider.getStateManager(id)); + validationContextFactory, stateManagerProvider.getStateManager(id), null); taskNode.setName(task.getClass().getSimpleName()); reportingTasks.put(id, taskNode); @@ -4601,6 +4619,7 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C return controllerServiceProvider.getControllerServiceIdentifiers(serviceType); } + /** * Captures snapshots of components' metrics */ http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/controller/reporting/ClusteredReportingTaskNode.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/controller/reporting/ClusteredReportingTaskNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/controller/reporting/ClusteredReportingTaskNode.java index a23cfdd..0adf41a 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/controller/reporting/ClusteredReportingTaskNode.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/controller/reporting/ClusteredReportingTaskNode.java @@ -21,6 +21,7 @@ import org.apache.nifi.components.state.StateManager; import org.apache.nifi.controller.ProcessScheduler; import org.apache.nifi.controller.ValidationContextFactory; import org.apache.nifi.controller.service.ControllerServiceProvider; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.reporting.BulletinRepository; import org.apache.nifi.reporting.EventAccess; import org.apache.nifi.reporting.ReportingContext; @@ -33,10 +34,12 @@ public class ClusteredReportingTaskNode extends AbstractReportingTaskNode { private final ControllerServiceProvider serviceProvider; private final StateManager stateManager; + public ClusteredReportingTaskNode(final ReportingTask reportingTask, final String id, final ProcessScheduler scheduler, - final EventAccess eventAccess, final BulletinRepository bulletinRepository, final ControllerServiceProvider serviceProvider, - final ValidationContextFactory validationContextFactory, final StateManager stateManager) { - super(reportingTask, id, serviceProvider, scheduler, validationContextFactory); + final EventAccess eventAccess, final BulletinRepository bulletinRepository, final ControllerServiceProvider serviceProvider, + final ValidationContextFactory validationContextFactory, final StateManager stateManager, + final VariableRegistry variableRegistry) { + super(reportingTask, id, serviceProvider, scheduler, validationContextFactory, variableRegistry); this.eventAccess = eventAccess; this.bulletinRepository = bulletinRepository; @@ -46,7 +49,7 @@ public class ClusteredReportingTaskNode extends AbstractReportingTaskNode { @Override public ReportingContext getReportingContext() { - return new ClusteredReportingContext(eventAccess, bulletinRepository, getProperties(), serviceProvider, stateManager); + return new ClusteredReportingContext(eventAccess, bulletinRepository, getProperties(), serviceProvider, stateManager, variableRegistry); } } http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java index a5f2ee0..7c8763b 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java @@ -140,6 +140,9 @@ import org.apache.nifi.provenance.ProvenanceEventRecord; import org.apache.nifi.provenance.ProvenanceEventRepository; import org.apache.nifi.provenance.ProvenanceEventType; import org.apache.nifi.provenance.StandardProvenanceEventRecord; +import org.apache.nifi.registry.VariableRegistry; +import org.apache.nifi.registry.VariableRegistryFactory; +import org.apache.nifi.registry.VariableRegistryUtils; import org.apache.nifi.remote.RemoteGroupPort; import org.apache.nifi.remote.RemoteResourceManager; import org.apache.nifi.remote.RemoteSiteListener; @@ -263,6 +266,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R private final StateManagerProvider stateManagerProvider; private final long systemStartTime = System.currentTimeMillis(); // time at which the node was started private final ConcurrentMap<String, ReportingTaskNode> reportingTasks = new ConcurrentHashMap<>(); + private final VariableRegistry variableRegistry; private volatile ZooKeeperStateServer zooKeeperStateServer; @@ -422,6 +426,8 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R bulletinRepository = new VolatileBulletinRepository(); nodeBulletinSubscriber = new AtomicReference<>(); + variableRegistry = createVariableRegistry(properties); + try { this.provenanceEventRepository = createProvenanceRepository(properties); this.provenanceEventRepository.initialize(createEventReporter(bulletinRepository)); @@ -436,21 +442,21 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R } try { - this.stateManagerProvider = StandardStateManagerProvider.create(properties); + this.stateManagerProvider = StandardStateManagerProvider.create(properties,variableRegistry); } catch (final IOException e) { throw new RuntimeException(e); } - processScheduler = new StandardProcessScheduler(this, this, encryptor, stateManagerProvider); + processScheduler = new StandardProcessScheduler(this, this, encryptor, stateManagerProvider, variableRegistry); eventDrivenWorkerQueue = new EventDrivenWorkerQueue(false, false, processScheduler); - controllerServiceProvider = new StandardControllerServiceProvider(processScheduler, bulletinRepository, stateManagerProvider); + controllerServiceProvider = new StandardControllerServiceProvider(processScheduler, bulletinRepository, stateManagerProvider, variableRegistry); final ProcessContextFactory contextFactory = new ProcessContextFactory(contentRepository, flowFileRepository, flowFileEventRepository, counterRepositoryRef.get(), provenanceEventRepository); processScheduler.setSchedulingAgent(SchedulingStrategy.EVENT_DRIVEN, new EventDrivenSchedulingAgent( - eventDrivenEngineRef.get(), this, stateManagerProvider, eventDrivenWorkerQueue, contextFactory, maxEventDrivenThreads.get(), encryptor)); + eventDrivenEngineRef.get(), this, stateManagerProvider, eventDrivenWorkerQueue, contextFactory, maxEventDrivenThreads.get(), encryptor, variableRegistry)); - final QuartzSchedulingAgent quartzSchedulingAgent = new QuartzSchedulingAgent(this, timerDrivenEngineRef.get(), contextFactory, encryptor); - final TimerDrivenSchedulingAgent timerDrivenAgent = new TimerDrivenSchedulingAgent(this, timerDrivenEngineRef.get(), contextFactory, encryptor); + final QuartzSchedulingAgent quartzSchedulingAgent = new QuartzSchedulingAgent(this, timerDrivenEngineRef.get(), contextFactory, encryptor, variableRegistry); + final TimerDrivenSchedulingAgent timerDrivenAgent = new TimerDrivenSchedulingAgent(this, timerDrivenEngineRef.get(), contextFactory, encryptor, variableRegistry); processScheduler.setSchedulingAgent(SchedulingStrategy.TIMER_DRIVEN, timerDrivenAgent); processScheduler.setSchedulingAgent(SchedulingStrategy.PRIMARY_NODE_ONLY, timerDrivenAgent); processScheduler.setSchedulingAgent(SchedulingStrategy.CRON_DRIVEN, quartzSchedulingAgent); @@ -491,7 +497,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R this.snippetManager = new SnippetManager(); - rootGroup = new StandardProcessGroup(UUID.randomUUID().toString(), this, processScheduler, properties, encryptor, this); + rootGroup = new StandardProcessGroup(UUID.randomUUID().toString(), this, processScheduler, properties, encryptor, this, variableRegistry); rootGroup.setName(DEFAULT_ROOT_GROUP_NAME); instanceId = UUID.randomUUID().toString(); @@ -538,8 +544,12 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R }, snapshotMillis, snapshotMillis, TimeUnit.MILLISECONDS); heartbeatBeanRef.set(new HeartbeatBean(rootGroup, false, false)); + + } + + private static FlowFileRepository createFlowFileRepository(final NiFiProperties properties, final ResourceClaimManager contentClaimManager) { final String implementationClassName = properties.getProperty(NiFiProperties.FLOWFILE_REPOSITORY_IMPLEMENTATION, DEFAULT_FLOWFILE_REPO_IMPLEMENTATION); if (implementationClassName == null) { @@ -583,6 +593,18 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R }; } + private static VariableRegistry createVariableRegistry(final NiFiProperties properties){ + VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry(); + try { + VariableRegistry customRegistry = VariableRegistryFactory.getPropertiesInstance(properties.getVariableRegistryPropertiesPaths()); + variableRegistry.addRegistry(customRegistry); + } catch (IOException ioe){ + LOG.error("Exception thrown while attempting to add properties to registry",ioe); + } + + return variableRegistry; + } + public void initializeFlow() throws IOException { writeLock.lock(); try { @@ -903,7 +925,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R * @throws NullPointerException if the argument is null */ public ProcessGroup createProcessGroup(final String id) { - return new StandardProcessGroup(requireNonNull(id).intern(), this, processScheduler, properties, encryptor, this); + return new StandardProcessGroup(requireNonNull(id).intern(), this, processScheduler, properties, encryptor, this, variableRegistry); } /** @@ -937,7 +959,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R public ProcessorNode createProcessor(final String type, String id, final boolean firstTimeAdded) throws ProcessorInstantiationException { id = id.intern(); final Processor processor = instantiateProcessor(type, id); - final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(controllerServiceProvider); + final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(controllerServiceProvider, variableRegistry); final ProcessorNode procNode = new StandardProcessorNode(processor, id, validationContextFactory, processScheduler, controllerServiceProvider); final LogRepository logRepository = LogRepositoryFactory.getRepository(id); @@ -1195,7 +1217,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R // invoke any methods annotated with @OnShutdown on Controller Services for (final ControllerServiceNode serviceNode : getAllControllerServices()) { try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { - final ConfigurationContext configContext = new StandardConfigurationContext(serviceNode, controllerServiceProvider, null); + final ConfigurationContext configContext = new StandardConfigurationContext(serviceNode, controllerServiceProvider, null, variableRegistry); ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnShutdown.class, serviceNode.getControllerServiceImplementation(), configContext); } } @@ -2666,8 +2688,8 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R } } - final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(controllerServiceProvider); - final ReportingTaskNode taskNode = new StandardReportingTaskNode(task, id, this, processScheduler, validationContextFactory); + final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(controllerServiceProvider, variableRegistry); + final ReportingTaskNode taskNode = new StandardReportingTaskNode(task, id, this, processScheduler, validationContextFactory, variableRegistry); taskNode.setName(task.getClass().getSimpleName()); if (firstTimeAdded) { http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/AbstractReportingTaskNode.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/AbstractReportingTaskNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/AbstractReportingTaskNode.java index c3eb0a0..31c2242 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/AbstractReportingTaskNode.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/AbstractReportingTaskNode.java @@ -36,6 +36,7 @@ import org.apache.nifi.controller.service.ControllerServiceNode; import org.apache.nifi.controller.service.ControllerServiceProvider; import org.apache.nifi.controller.service.StandardConfigurationContext; import org.apache.nifi.nar.NarCloseable; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.reporting.ReportingTask; import org.apache.nifi.scheduling.SchedulingStrategy; import org.apache.nifi.util.FormatUtils; @@ -47,19 +48,23 @@ public abstract class AbstractReportingTaskNode extends AbstractConfiguredCompon private final ProcessScheduler processScheduler; private final ControllerServiceLookup serviceLookup; + private final AtomicReference<SchedulingStrategy> schedulingStrategy = new AtomicReference<>(SchedulingStrategy.TIMER_DRIVEN); private final AtomicReference<String> schedulingPeriod = new AtomicReference<>("5 mins"); private volatile String comment; private volatile ScheduledState scheduledState = ScheduledState.STOPPED; + protected final VariableRegistry variableRegistry; + public AbstractReportingTaskNode(final ReportingTask reportingTask, final String id, - final ControllerServiceProvider controllerServiceProvider, final ProcessScheduler processScheduler, - final ValidationContextFactory validationContextFactory) { + final ControllerServiceProvider controllerServiceProvider, final ProcessScheduler processScheduler, + final ValidationContextFactory validationContextFactory, final VariableRegistry variableRegistry) { super(reportingTask, id, validationContextFactory, controllerServiceProvider); this.reportingTask = reportingTask; this.processScheduler = processScheduler; this.serviceLookup = controllerServiceProvider; + this.variableRegistry = variableRegistry; } @Override @@ -104,7 +109,7 @@ public abstract class AbstractReportingTaskNode extends AbstractConfiguredCompon @Override public ConfigurationContext getConfigurationContext() { - return new StandardConfigurationContext(this, serviceLookup, getSchedulingPeriod()); + return new StandardConfigurationContext(this, serviceLookup, getSchedulingPeriod(), variableRegistry); } @Override @@ -146,7 +151,7 @@ public abstract class AbstractReportingTaskNode extends AbstractConfiguredCompon // We need to invoke any method annotation with the OnConfigured annotation in order to // maintain backward compatibility. This will be removed when we remove the old, deprecated annotations. try (final NarCloseable x = NarCloseable.withNarLoader()) { - final ConfigurationContext configContext = new StandardConfigurationContext(this, serviceLookup, getSchedulingPeriod()); + final ConfigurationContext configContext = new StandardConfigurationContext(this, serviceLookup, getSchedulingPeriod(), variableRegistry); ReflectionUtils.invokeMethodsWithAnnotation(OnConfigured.class, reportingTask, configContext); } catch (final Exception e) { throw new ComponentLifeCycleException("Failed to invoke On-Configured Lifecycle methods of " + reportingTask, e);
