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/StandardReportingContext.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingContext.java index 11d1b51..a0a614c 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingContext.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingContext.java @@ -34,6 +34,7 @@ import org.apache.nifi.controller.FlowController; import org.apache.nifi.controller.service.ControllerServiceProvider; import org.apache.nifi.events.BulletinFactory; import org.apache.nifi.groups.ProcessGroup; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.reporting.Bulletin; import org.apache.nifi.reporting.BulletinRepository; import org.apache.nifi.reporting.EventAccess; @@ -50,16 +51,18 @@ public class StandardReportingContext implements ReportingContext, ControllerSer private final ControllerServiceProvider serviceProvider; private final Map<PropertyDescriptor, String> properties; private final Map<PropertyDescriptor, PreparedQuery> preparedQueries; + private final VariableRegistry variableRegistry; public StandardReportingContext(final FlowController flowController, final BulletinRepository bulletinRepository, - final Map<PropertyDescriptor, String> properties, final ControllerServiceProvider serviceProvider, final ReportingTask reportingTask) { + final Map<PropertyDescriptor, String> properties, final ControllerServiceProvider serviceProvider, final ReportingTask reportingTask, + final VariableRegistry variableRegistry) { this.flowController = flowController; this.eventAccess = flowController; this.bulletinRepository = bulletinRepository; this.properties = Collections.unmodifiableMap(properties); this.serviceProvider = serviceProvider; this.reportingTask = reportingTask; - + this.variableRegistry = variableRegistry; preparedQueries = new HashMap<>(); for (final Map.Entry<PropertyDescriptor, String> entry : properties.entrySet()) { final PropertyDescriptor desc = entry.getKey(); @@ -106,7 +109,7 @@ public class StandardReportingContext implements ReportingContext, ControllerSer @Override public PropertyValue getProperty(final PropertyDescriptor property) { final String configuredValue = properties.get(property); - return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, this, preparedQueries.get(property)); + return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, this, preparedQueries.get(property), variableRegistry); } @Override @@ -148,4 +151,5 @@ public class StandardReportingContext implements ReportingContext, ControllerSer public StateManager getStateManager() { return flowController.getStateManagerProvider().getStateManager(reportingTask.getIdentifier()); } + }
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/StandardReportingInitializationContext.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingInitializationContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingInitializationContext.java index 0131a95..312d99c 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingInitializationContext.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingInitializationContext.java @@ -112,4 +112,5 @@ public class StandardReportingInitializationContext implements ReportingInitiali public ComponentLog getLogger() { return logger; } + } 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/StandardReportingTaskNode.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingTaskNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingTaskNode.java index 1a40e8a..c5ce875 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingTaskNode.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingTaskNode.java @@ -20,6 +20,7 @@ import org.apache.nifi.controller.FlowController; import org.apache.nifi.controller.ProcessScheduler; import org.apache.nifi.controller.ReportingTaskNode; import org.apache.nifi.controller.ValidationContextFactory; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.reporting.ReportingContext; import org.apache.nifi.reporting.ReportingTask; @@ -28,13 +29,14 @@ public class StandardReportingTaskNode extends AbstractReportingTaskNode impleme private final FlowController flowController; public StandardReportingTaskNode(final ReportingTask reportingTask, final String id, final FlowController controller, - final ProcessScheduler processScheduler, final ValidationContextFactory validationContextFactory) { - super(reportingTask, id, controller, processScheduler, validationContextFactory); + final ProcessScheduler processScheduler, final ValidationContextFactory validationContextFactory, + final VariableRegistry variableRegistry) { + super(reportingTask, id, controller, processScheduler, validationContextFactory, variableRegistry); this.flowController = controller; } @Override public ReportingContext getReportingContext() { - return new StandardReportingContext(flowController, flowController.getBulletinRepository(), getProperties(), flowController, getReportingTask()); + return new StandardReportingContext(flowController, flowController.getBulletinRepository(), getProperties(), flowController, getReportingTask(), 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/scheduling/EventDrivenSchedulingAgent.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/EventDrivenSchedulingAgent.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/EventDrivenSchedulingAgent.java index 228af70..af31722 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/EventDrivenSchedulingAgent.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/EventDrivenSchedulingAgent.java @@ -45,6 +45,7 @@ import org.apache.nifi.processor.ProcessSessionFactory; import org.apache.nifi.processor.SimpleProcessLogger; import org.apache.nifi.processor.StandardProcessContext; import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.util.Connectables; import org.apache.nifi.util.FormatUtils; import org.apache.nifi.util.ReflectionUtils; @@ -61,6 +62,7 @@ public class EventDrivenSchedulingAgent extends AbstractSchedulingAgent { private final ProcessContextFactory contextFactory; private final AtomicInteger maxThreadCount; private final StringEncryptor encryptor; + private final VariableRegistry variableRegistry; private volatile String adminYieldDuration = "1 sec"; @@ -68,7 +70,8 @@ public class EventDrivenSchedulingAgent extends AbstractSchedulingAgent { private final ConcurrentMap<Connectable, ScheduleState> scheduleStates = new ConcurrentHashMap<>(); public EventDrivenSchedulingAgent(final FlowEngine flowEngine, final ControllerServiceProvider serviceProvider, final StateManagerProvider stateManagerProvider, - final EventDrivenWorkerQueue workerQueue, final ProcessContextFactory contextFactory, final int maxThreadCount, final StringEncryptor encryptor) { + final EventDrivenWorkerQueue workerQueue, final ProcessContextFactory contextFactory, final int maxThreadCount, final StringEncryptor encryptor, + final VariableRegistry variableRegistry) { super(flowEngine); this.serviceProvider = serviceProvider; this.stateManagerProvider = stateManagerProvider; @@ -76,6 +79,7 @@ public class EventDrivenSchedulingAgent extends AbstractSchedulingAgent { this.contextFactory = contextFactory; this.maxThreadCount = new AtomicInteger(maxThreadCount); this.encryptor = encryptor; + this.variableRegistry = variableRegistry; for (int i = 0; i < maxThreadCount; i++) { final Runnable eventDrivenTask = new EventDrivenTask(workerQueue); @@ -185,7 +189,7 @@ public class EventDrivenSchedulingAgent extends AbstractSchedulingAgent { if (connectable instanceof ProcessorNode) { final ProcessorNode procNode = (ProcessorNode) connectable; final StandardProcessContext standardProcessContext = new StandardProcessContext(procNode, serviceProvider, - encryptor, getStateManager(connectable.getIdentifier())); + encryptor, getStateManager(connectable.getIdentifier()), variableRegistry); final long runNanos = procNode.getRunDuration(TimeUnit.NANOSECONDS); final ProcessSessionFactory sessionFactory; 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/scheduling/QuartzSchedulingAgent.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/QuartzSchedulingAgent.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/QuartzSchedulingAgent.java index 3f19d28..34e7989 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/QuartzSchedulingAgent.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/QuartzSchedulingAgent.java @@ -38,6 +38,7 @@ import org.apache.nifi.encrypt.StringEncryptor; import org.apache.nifi.engine.FlowEngine; import org.apache.nifi.processor.StandardProcessContext; import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.util.FormatUtils; import org.quartz.CronExpression; import org.slf4j.Logger; @@ -50,15 +51,18 @@ public class QuartzSchedulingAgent extends AbstractSchedulingAgent { private final FlowController flowController; private final ProcessContextFactory contextFactory; private final StringEncryptor encryptor; + private final VariableRegistry variableRegistry; private volatile String adminYieldDuration = "1 sec"; private final Map<Object, List<AtomicBoolean>> canceledTriggers = new HashMap<>(); - public QuartzSchedulingAgent(final FlowController flowController, final FlowEngine flowEngine, final ProcessContextFactory contextFactory, final StringEncryptor enryptor) { + public QuartzSchedulingAgent(final FlowController flowController, final FlowEngine flowEngine, final ProcessContextFactory contextFactory, final StringEncryptor enryptor, + final VariableRegistry variableRegistry) { super(flowEngine); this.flowController = flowController; this.contextFactory = contextFactory; this.encryptor = enryptor; + this.variableRegistry = variableRegistry; } private StateManager getStateManager(final String componentId) { @@ -141,7 +145,7 @@ public class QuartzSchedulingAgent extends AbstractSchedulingAgent { if (connectable.getConnectableType() == ConnectableType.PROCESSOR) { final ProcessorNode procNode = (ProcessorNode) connectable; - final StandardProcessContext standardProcContext = new StandardProcessContext(procNode, flowController, encryptor, getStateManager(connectable.getIdentifier())); + final StandardProcessContext standardProcContext = new StandardProcessContext(procNode, flowController, encryptor, getStateManager(connectable.getIdentifier()), variableRegistry); ContinuallyRunProcessorTask runnableTask = new ContinuallyRunProcessorTask(this, procNode, flowController, contextFactory, scheduleState, standardProcContext); continuallyRunTask = runnableTask; } else { 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/scheduling/StandardProcessScheduler.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java index e8b2146..86dac7b 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java @@ -53,6 +53,7 @@ import org.apache.nifi.nar.NarCloseable; import org.apache.nifi.processor.Processor; import org.apache.nifi.processor.SimpleProcessLogger; import org.apache.nifi.processor.StandardProcessContext; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.reporting.ReportingTask; import org.apache.nifi.scheduling.SchedulingStrategy; import org.apache.nifi.util.FormatUtils; @@ -83,13 +84,15 @@ public final class StandardProcessScheduler implements ProcessScheduler { private final ScheduledExecutorService componentMonitoringThreadPool = new FlowEngine(8, "StandardProcessScheduler", true); private final StringEncryptor encryptor; + private final VariableRegistry variableRegistry; public StandardProcessScheduler(final Heartbeater heartbeater, final ControllerServiceProvider controllerServiceProvider, final StringEncryptor encryptor, - final StateManagerProvider stateManagerProvider) { + final StateManagerProvider stateManagerProvider, final VariableRegistry variableRegistry) { this.heartbeater = heartbeater; this.controllerServiceProvider = controllerServiceProvider; this.encryptor = encryptor; this.stateManagerProvider = stateManagerProvider; + this.variableRegistry = variableRegistry; administrativeYieldDuration = NiFiProperties.getInstance().getAdministrativeYieldDuration(); administrativeYieldMillis = FormatUtils.getTimeDuration(administrativeYieldDuration, TimeUnit.MILLISECONDS); @@ -295,7 +298,7 @@ public final class StandardProcessScheduler implements ProcessScheduler { @Override public synchronized void startProcessor(final ProcessorNode procNode) { StandardProcessContext processContext = new StandardProcessContext(procNode, this.controllerServiceProvider, - this.encryptor, getStateManager(procNode.getIdentifier())); + this.encryptor, getStateManager(procNode.getIdentifier()), variableRegistry); final ScheduleState scheduleState = getScheduleState(requireNonNull(procNode)); SchedulingAgentCallback callback = new SchedulingAgentCallback() { @@ -330,7 +333,7 @@ public final class StandardProcessScheduler implements ProcessScheduler { @Override public synchronized void stopProcessor(final ProcessorNode procNode) { StandardProcessContext processContext = new StandardProcessContext(procNode, this.controllerServiceProvider, - this.encryptor, getStateManager(procNode.getIdentifier())); + this.encryptor, getStateManager(procNode.getIdentifier()), variableRegistry); final ScheduleState state = getScheduleState(procNode); procNode.stop(this.componentLifeCycleThreadPool, processContext, new Callable<Boolean>() { 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/scheduling/TimerDrivenSchedulingAgent.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java index 0436e21..f94beff 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java @@ -37,6 +37,7 @@ import org.apache.nifi.engine.FlowEngine; import org.apache.nifi.processor.ProcessContext; import org.apache.nifi.processor.StandardProcessContext; import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.util.FormatUtils; import org.apache.nifi.util.NiFiProperties; import org.slf4j.Logger; @@ -50,14 +51,17 @@ public class TimerDrivenSchedulingAgent extends AbstractSchedulingAgent { private final FlowController flowController; private final ProcessContextFactory contextFactory; private final StringEncryptor encryptor; + private final VariableRegistry variableRegistry; private volatile String adminYieldDuration = "1 sec"; - public TimerDrivenSchedulingAgent(final FlowController flowController, final FlowEngine flowEngine, final ProcessContextFactory contextFactory, final StringEncryptor encryptor) { + public TimerDrivenSchedulingAgent(final FlowController flowController, final FlowEngine flowEngine, final ProcessContextFactory contextFactory, final StringEncryptor encryptor, + final VariableRegistry variableRegistry) { super(flowEngine); this.flowController = flowController; this.contextFactory = contextFactory; this.encryptor = encryptor; + this.variableRegistry = variableRegistry; final String boredYieldDuration = NiFiProperties.getInstance().getBoredYieldDuration(); try { @@ -100,7 +104,7 @@ public class TimerDrivenSchedulingAgent extends AbstractSchedulingAgent { // Determine the task to run and create it. if (connectable.getConnectableType() == ConnectableType.PROCESSOR) { final ProcessorNode procNode = (ProcessorNode) connectable; - final StandardProcessContext standardProcContext = new StandardProcessContext(procNode, flowController, encryptor, getStateManager(connectable.getIdentifier())); + final StandardProcessContext standardProcContext = new StandardProcessContext(procNode, flowController, encryptor, getStateManager(connectable.getIdentifier()), variableRegistry); final ContinuallyRunProcessorTask runnableTask = new ContinuallyRunProcessorTask(this, procNode, flowController, contextFactory, scheduleState, standardProcContext); 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/service/StandardConfigurationContext.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardConfigurationContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardConfigurationContext.java index d57e61f..61db819 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardConfigurationContext.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardConfigurationContext.java @@ -28,6 +28,7 @@ import org.apache.nifi.components.PropertyValue; import org.apache.nifi.controller.ConfigurationContext; import org.apache.nifi.controller.ConfiguredComponent; import org.apache.nifi.controller.ControllerServiceLookup; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.util.FormatUtils; public class StandardConfigurationContext implements ConfigurationContext { @@ -35,13 +36,17 @@ public class StandardConfigurationContext implements ConfigurationContext { private final ConfiguredComponent component; private final ControllerServiceLookup serviceLookup; private final Map<PropertyDescriptor, PreparedQuery> preparedQueries; + private final VariableRegistry variableRegistry; private final String schedulingPeriod; private final Long schedulingNanos; - public StandardConfigurationContext(final ConfiguredComponent component, final ControllerServiceLookup serviceLookup, final String schedulingPeriod) { + public StandardConfigurationContext(final ConfiguredComponent component, final ControllerServiceLookup serviceLookup, final String schedulingPeriod, + final VariableRegistry variableRegistry) { this.component = component; this.serviceLookup = serviceLookup; this.schedulingPeriod = schedulingPeriod; + this.variableRegistry = variableRegistry; + if (schedulingPeriod == null) { schedulingNanos = null; } else { @@ -68,7 +73,7 @@ public class StandardConfigurationContext implements ConfigurationContext { @Override public PropertyValue getProperty(final PropertyDescriptor property) { final String configuredValue = component.getProperty(property); - return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, serviceLookup, preparedQueries.get(property)); + return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, serviceLookup, 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-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java index 482dabf..f861f6b 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java @@ -87,4 +87,5 @@ public class StandardControllerServiceInitializationContext implements Controlle public StateManager getStateManager() { return stateManager; } + } 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/service/StandardControllerServiceNode.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java index 3f24ff1..83c35ad 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java @@ -43,6 +43,7 @@ import org.apache.nifi.controller.exception.ComponentLifeCycleException; import org.apache.nifi.logging.ComponentLog; import org.apache.nifi.nar.NarCloseable; import org.apache.nifi.processor.SimpleProcessLogger; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.util.ReflectionUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -54,7 +55,7 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i private final ControllerService proxedControllerService; private final ControllerService implementation; private final ControllerServiceProvider serviceProvider; - + private final VariableRegistry variableRegistry; private final AtomicReference<ControllerServiceState> stateRef = new AtomicReference<>(ControllerServiceState.DISABLED); private final ReadWriteLock rwLock = new ReentrantReadWriteLock(); @@ -67,12 +68,15 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i private final AtomicBoolean active; public StandardControllerServiceNode(final ControllerService proxiedControllerService, final ControllerService implementation, final String id, - final ValidationContextFactory validationContextFactory, final ControllerServiceProvider serviceProvider) { + final ValidationContextFactory validationContextFactory, final ControllerServiceProvider serviceProvider, + final VariableRegistry variableRegistry) { super(implementation, id, validationContextFactory, serviceProvider); this.proxedControllerService = proxiedControllerService; this.implementation = implementation; this.serviceProvider = serviceProvider; this.active = new AtomicBoolean(); + this.variableRegistry = variableRegistry; + } @Override @@ -141,7 +145,7 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i @SuppressWarnings("deprecation") private void onConfigured() { try (final NarCloseable x = NarCloseable.withNarLoader()) { - final ConfigurationContext configContext = new StandardConfigurationContext(this, serviceProvider, null); + final ConfigurationContext configContext = new StandardConfigurationContext(this, serviceProvider, null, variableRegistry); ReflectionUtils.invokeMethodsWithAnnotation(OnConfigured.class, implementation, configContext); } catch (final Exception e) { throw new ComponentLifeCycleException("Failed to invoke On-Configured Lifecycle methods of " + implementation, e); @@ -277,7 +281,7 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i final Heartbeater heartbeater) { if (this.stateRef.compareAndSet(ControllerServiceState.DISABLED, ControllerServiceState.ENABLING)) { this.active.set(true); - final ConfigurationContext configContext = new StandardConfigurationContext(this, this.serviceProvider, null); + final ConfigurationContext configContext = new StandardConfigurationContext(this, this.serviceProvider, null, variableRegistry); scheduler.execute(new Runnable() { @Override public void run() { @@ -342,7 +346,7 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i } if (this.stateRef.compareAndSet(ControllerServiceState.ENABLED, ControllerServiceState.DISABLING)) { - final ConfigurationContext configContext = new StandardConfigurationContext(this, this.serviceProvider, null); + final ConfigurationContext configContext = new StandardConfigurationContext(this, this.serviceProvider, null, variableRegistry); scheduler.execute(new Runnable() { @Override public void run() { 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/service/StandardControllerServiceProvider.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java index 77dc87e..d0ad127 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java @@ -56,6 +56,7 @@ import org.apache.nifi.nar.ExtensionManager; import org.apache.nifi.nar.NarCloseable; import org.apache.nifi.processor.SimpleProcessLogger; import org.apache.nifi.processor.StandardValidationContextFactory; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.reporting.BulletinRepository; import org.apache.nifi.reporting.Severity; import org.apache.nifi.util.ObjectHolder; @@ -72,6 +73,7 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi private static final Set<Method> validDisabledMethods; private final BulletinRepository bulletinRepo; private final StateManagerProvider stateManagerProvider; + private final VariableRegistry variableRegistry; static { // methods that are okay to be called when the service is disabled. @@ -85,13 +87,15 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi validDisabledMethods = Collections.unmodifiableSet(validMethods); } - public StandardControllerServiceProvider(final ProcessScheduler scheduler, final BulletinRepository bulletinRepo, final StateManagerProvider stateManagerProvider) { + public StandardControllerServiceProvider(final ProcessScheduler scheduler, final BulletinRepository bulletinRepo, final StateManagerProvider stateManagerProvider, + final VariableRegistry variableRegistry) { // the following 2 maps must be updated atomically, but we do not lock around them because they are modified // only in the createControllerService method, and both are modified before the method returns this.controllerServices = new ConcurrentHashMap<>(); this.processScheduler = scheduler; this.bulletinRepo = bulletinRepo; this.stateManagerProvider = stateManagerProvider; + this.variableRegistry = variableRegistry; } private Class<?>[] getInterfaces(final Class<?> cls) { @@ -181,9 +185,9 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi final ComponentLog serviceLogger = new SimpleProcessLogger(id, originalService); originalService.initialize(new StandardControllerServiceInitializationContext(id, serviceLogger, this, getStateManager(id))); - final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(this); + final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(this, variableRegistry); - final ControllerServiceNode serviceNode = new StandardControllerServiceNode(proxiedService, originalService, id, validationContextFactory, this); + final ControllerServiceNode serviceNode = new StandardControllerServiceNode(proxiedService, originalService, id, validationContextFactory, this, variableRegistry); serviceNodeHolder.set(serviceNode); serviceNode.setName(rawClass.getSimpleName()); @@ -481,7 +485,7 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi serviceNode.verifyCanDelete(); try (final NarCloseable x = NarCloseable.withNarLoader()) { - final ConfigurationContext configurationContext = new StandardConfigurationContext(serviceNode, this, null); + final ConfigurationContext configurationContext = new StandardConfigurationContext(serviceNode, this, null, variableRegistry); ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnRemoved.class, serviceNode.getControllerServiceImplementation(), configurationContext); } @@ -615,4 +619,5 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi public void verifyCanStopReferencingComponents(final ControllerServiceNode serviceNode) { // we can always stop referencing components } + } 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/state/manager/StandardStateManagerProvider.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/manager/StandardStateManagerProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/manager/StandardStateManagerProvider.java index ef46f55..385f93a 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/manager/StandardStateManagerProvider.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/manager/StandardStateManagerProvider.java @@ -46,11 +46,12 @@ import org.apache.nifi.controller.state.config.StateProviderConfiguration; import org.apache.nifi.framework.security.util.SslContextFactory; import org.apache.nifi.nar.ExtensionManager; import org.apache.nifi.processor.StandardValidationContext; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.util.NiFiProperties; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public class StandardStateManagerProvider implements StateManagerProvider { +public class StandardStateManagerProvider implements StateManagerProvider{ private static final Logger logger = LoggerFactory.getLogger(StandardStateManagerProvider.class); private final ConcurrentMap<String, StateManager> stateManagers = new ConcurrentHashMap<>(); @@ -62,12 +63,12 @@ public class StandardStateManagerProvider implements StateManagerProvider { this.clusterStateProvider = clusterStateProvider; } - public static StateManagerProvider create(final NiFiProperties properties) throws ConfigParseException, IOException { - final StateProvider localProvider = createLocalStateProvider(properties); + public static StateManagerProvider create(final NiFiProperties properties, final VariableRegistry variableRegistry) throws ConfigParseException, IOException { + final StateProvider localProvider = createLocalStateProvider(properties,variableRegistry); final StateProvider clusterProvider; if (properties.isNode()) { - clusterProvider = createClusteredStateProvider(properties); + clusterProvider = createClusteredStateProvider(properties,variableRegistry); } else { clusterProvider = null; } @@ -75,25 +76,27 @@ public class StandardStateManagerProvider implements StateManagerProvider { return new StandardStateManagerProvider(localProvider, clusterProvider); } - private static StateProvider createLocalStateProvider(final NiFiProperties properties) throws IOException, ConfigParseException { + private static StateProvider createLocalStateProvider(final NiFiProperties properties, final VariableRegistry variableRegistry) throws IOException, ConfigParseException { final File configFile = properties.getStateManagementConfigFile(); - return createStateProvider(configFile, Scope.LOCAL, properties); + return createStateProvider(configFile, Scope.LOCAL, properties, variableRegistry); } - private static StateProvider createClusteredStateProvider(final NiFiProperties properties) throws IOException, ConfigParseException { + private static StateProvider createClusteredStateProvider(final NiFiProperties properties, final VariableRegistry variableRegistry) throws IOException, ConfigParseException { final File configFile = properties.getStateManagementConfigFile(); - return createStateProvider(configFile, Scope.CLUSTER, properties); + return createStateProvider(configFile, Scope.CLUSTER, properties, variableRegistry); } - private static StateProvider createStateProvider(final File configFile, final Scope scope, final NiFiProperties properties) throws ConfigParseException, IOException { + private static StateProvider createStateProvider(final File configFile, final Scope scope, final NiFiProperties properties, + final VariableRegistry variableRegistry) throws ConfigParseException, IOException { final String providerId; final String providerIdPropertyName; final String providerDescription; final String providerXmlElementName; final String oppositeScopeXmlElementName; + switch (scope) { case CLUSTER: providerId = properties.getClusterStateProviderId(); @@ -166,17 +169,18 @@ public class StandardStateManagerProvider implements StateManagerProvider { + " is configured to use scope " + scope); } + //create variable registry final Map<PropertyDescriptor, PropertyValue> propertyMap = new HashMap<>(); final Map<PropertyDescriptor, String> propertyStringMap = new HashMap<>(); for (final PropertyDescriptor descriptor : provider.getPropertyDescriptors()) { - propertyMap.put(descriptor, new StandardPropertyValue(descriptor.getDefaultValue(), null)); + propertyMap.put(descriptor, new StandardPropertyValue(descriptor.getDefaultValue(),null, variableRegistry)); propertyStringMap.put(descriptor, descriptor.getDefaultValue()); } for (final Map.Entry<String, String> entry : providerConfig.getProperties().entrySet()) { final PropertyDescriptor descriptor = provider.getPropertyDescriptor(entry.getKey()); propertyStringMap.put(descriptor, entry.getValue()); - propertyMap.put(descriptor, new StandardPropertyValue(entry.getValue(), null)); + propertyMap.put(descriptor, new StandardPropertyValue(entry.getValue(),null, variableRegistry)); } final SSLContext sslContext = SslContextFactory.createSslContext(properties, false); @@ -186,7 +190,7 @@ public class StandardStateManagerProvider implements StateManagerProvider { provider.initialize(initContext); } - final ValidationContext validationContext = new StandardValidationContext(null, propertyStringMap, null); + final ValidationContext validationContext = new StandardValidationContext(null, propertyStringMap, null, variableRegistry); final Collection<ValidationResult> results = provider.validate(validationContext); final StringBuilder validationFailures = new StringBuilder(); 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/groups/StandardProcessGroup.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java index 4646d55..e9032eb 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java @@ -59,6 +59,7 @@ import org.apache.nifi.encrypt.StringEncryptor; import org.apache.nifi.logging.LogRepositoryFactory; import org.apache.nifi.nar.NarCloseable; import org.apache.nifi.processor.StandardProcessContext; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.remote.RemoteGroupPort; import org.apache.nifi.remote.RootGroupPort; import org.apache.nifi.util.NiFiProperties; @@ -87,6 +88,7 @@ public final class StandardProcessGroup implements ProcessGroup { private final Map<String, ProcessorNode> processors = new HashMap<>(); private final Map<String, Funnel> funnels = new HashMap<>(); private final StringEncryptor encryptor; + private final VariableRegistry variableRegistry; private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock(); private final Lock readLock = rwLock.readLock(); @@ -95,7 +97,8 @@ public final class StandardProcessGroup implements ProcessGroup { private static final Logger LOG = LoggerFactory.getLogger(StandardProcessGroup.class); public StandardProcessGroup(final String id, final ControllerServiceProvider serviceProvider, final StandardProcessScheduler scheduler, - final NiFiProperties nifiProps, final StringEncryptor encryptor, final FlowController flowController) { + final NiFiProperties nifiProps, final StringEncryptor encryptor, final FlowController flowController, + final VariableRegistry variableRegistry) { this.id = id; this.controllerServiceProvider = serviceProvider; this.parent = new AtomicReference<>(); @@ -103,6 +106,7 @@ public final class StandardProcessGroup implements ProcessGroup { this.comments = new AtomicReference<>(""); this.encryptor = encryptor; this.flowController = flowController; + this.variableRegistry = variableRegistry; name = new AtomicReference<>(); position = new AtomicReference<>(new Position(0D, 0D)); @@ -342,7 +346,7 @@ public final class StandardProcessGroup implements ProcessGroup { private void shutdown(final ProcessGroup procGroup) { for (final ProcessorNode node : procGroup.getProcessors()) { try (final NarCloseable x = NarCloseable.withNarLoader()) { - final StandardProcessContext processContext = new StandardProcessContext(node, controllerServiceProvider, encryptor, getStateManager(node.getIdentifier())); + final StandardProcessContext processContext = new StandardProcessContext(node, controllerServiceProvider, encryptor, getStateManager(node.getIdentifier()), variableRegistry); ReflectionUtils.quietlyInvokeMethodsWithAnnotations(OnShutdown.class, org.apache.nifi.processor.annotation.OnShutdown.class, node.getProcessor(), processContext); } } @@ -708,7 +712,7 @@ public final class StandardProcessGroup implements ProcessGroup { } try (final NarCloseable x = NarCloseable.withNarLoader()) { - final StandardProcessContext processContext = new StandardProcessContext(processor, controllerServiceProvider, encryptor, getStateManager(processor.getIdentifier())); + final StandardProcessContext processContext = new StandardProcessContext(processor, controllerServiceProvider, encryptor, getStateManager(processor.getIdentifier()), variableRegistry); ReflectionUtils.quietlyInvokeMethodsWithAnnotations(OnRemoved.class, org.apache.nifi.processor.annotation.OnRemoved.class, processor.getProcessor(), processContext); } catch (final Exception e) { throw new ComponentLifeCycleException("Failed to invoke 'OnRemoved' methods of " + processor, e); @@ -2097,4 +2101,5 @@ public final class StandardProcessGroup implements ProcessGroup { readLock.unlock(); } } + } 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/processor/StandardProcessContext.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessContext.java index dae7d0c..09c68c2 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessContext.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessContext.java @@ -36,6 +36,7 @@ import org.apache.nifi.controller.ControllerServiceLookup; import org.apache.nifi.controller.ProcessorNode; import org.apache.nifi.controller.service.ControllerServiceProvider; import org.apache.nifi.encrypt.StringEncryptor; +import org.apache.nifi.registry.VariableRegistry; import org.apache.nifi.util.Connectables; public class StandardProcessContext implements ProcessContext, ControllerServiceLookup { @@ -45,12 +46,15 @@ public class StandardProcessContext implements ProcessContext, ControllerService private final Map<PropertyDescriptor, PreparedQuery> preparedQueries; private final StringEncryptor encryptor; private final StateManager stateManager; + private final VariableRegistry variableRegistry; - public StandardProcessContext(final ProcessorNode processorNode, final ControllerServiceProvider controllerServiceProvider, final StringEncryptor encryptor, final StateManager stateManager) { + public StandardProcessContext(final ProcessorNode processorNode, final ControllerServiceProvider controllerServiceProvider, final StringEncryptor encryptor, final StateManager stateManager, + final VariableRegistry variableRegistry) { this.procNode = processorNode; this.controllerServiceProvider = controllerServiceProvider; this.encryptor = encryptor; this.stateManager = stateManager; + this.variableRegistry = variableRegistry; preparedQueries = new HashMap<>(); for (final Map.Entry<PropertyDescriptor, String> entry : procNode.getProperties().entrySet()) { @@ -86,12 +90,12 @@ public class StandardProcessContext implements ProcessContext, ControllerService final String setPropertyValue = procNode.getProperty(descriptor); final String propValue = (setPropertyValue == null) ? descriptor.getDefaultValue() : setPropertyValue; - return new StandardPropertyValue(propValue, this, preparedQueries.get(descriptor)); + return new StandardPropertyValue(propValue, this, preparedQueries.get(descriptor), variableRegistry); } @Override public PropertyValue newPropertyValue(final String rawValue) { - return new StandardPropertyValue(rawValue, this, Query.prepare(rawValue)); + return new StandardPropertyValue(rawValue, this, Query.prepare(rawValue), variableRegistry); } @Override @@ -221,4 +225,5 @@ public class StandardProcessContext implements ProcessContext, ControllerService public String getName() { return procNode.getName(); } + } 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/processor/StandardValidationContext.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContext.java index c24e146..5489221 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContext.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContext.java @@ -35,6 +35,7 @@ import org.apache.nifi.controller.ControllerServiceLookup; import org.apache.nifi.controller.service.ControllerServiceNode; import org.apache.nifi.controller.service.ControllerServiceProvider; import org.apache.nifi.expression.ExpressionLanguageCompiler; +import org.apache.nifi.registry.VariableRegistry; public class StandardValidationContext implements ValidationContext { @@ -44,20 +45,23 @@ public class StandardValidationContext implements ValidationContext { private final Map<String, Boolean> expressionLanguageSupported; private final String annotationData; private final Set<String> serviceIdentifiersToNotValidate; + private final VariableRegistry variableRegistry; - public StandardValidationContext(final ControllerServiceProvider controllerServiceProvider, final Map<PropertyDescriptor, String> properties, final String annotationData) { - this(controllerServiceProvider, Collections.<String>emptySet(), properties, annotationData); + public StandardValidationContext(final ControllerServiceProvider controllerServiceProvider, final Map<PropertyDescriptor, String> properties, final String annotationData, + final VariableRegistry variableRegistry) { + this(controllerServiceProvider, Collections.<String>emptySet(), properties, annotationData, variableRegistry); } public StandardValidationContext( final ControllerServiceProvider controllerServiceProvider, final Set<String> serviceIdentifiersToNotValidate, final Map<PropertyDescriptor, String> properties, - final String annotationData) { + final String annotationData, VariableRegistry variableRegistry) { this.controllerServiceProvider = controllerServiceProvider; this.properties = new HashMap<>(properties); this.annotationData = annotationData; this.serviceIdentifiersToNotValidate = serviceIdentifiersToNotValidate; + this.variableRegistry = variableRegistry; preparedQueries = new HashMap<>(properties.size()); for (final Map.Entry<PropertyDescriptor, String> entry : properties.entrySet()) { @@ -79,24 +83,24 @@ public class StandardValidationContext implements ValidationContext { @Override public PropertyValue newPropertyValue(final String rawValue) { - return new StandardPropertyValue(rawValue, controllerServiceProvider, Query.prepare(rawValue)); + return new StandardPropertyValue(rawValue, controllerServiceProvider, Query.prepare(rawValue), variableRegistry); } @Override public ExpressionLanguageCompiler newExpressionLanguageCompiler() { - return new StandardExpressionLanguageCompiler(); + return new StandardExpressionLanguageCompiler(variableRegistry); } @Override public ValidationContext getControllerServiceValidationContext(final ControllerService controllerService) { final ControllerServiceNode serviceNode = controllerServiceProvider.getControllerServiceNode(controllerService.getIdentifier()); - return new StandardValidationContext(controllerServiceProvider, serviceNode.getProperties(), serviceNode.getAnnotationData()); + return new StandardValidationContext(controllerServiceProvider, serviceNode.getProperties(), serviceNode.getAnnotationData(), variableRegistry); } @Override public PropertyValue getProperty(final PropertyDescriptor property) { final String configuredValue = properties.get(property); - return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, controllerServiceProvider, preparedQueries.get(property)); + return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, controllerServiceProvider, 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-core/src/main/java/org/apache/nifi/processor/StandardValidationContextFactory.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContextFactory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContextFactory.java index c3df987..e6b21e0 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContextFactory.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContextFactory.java @@ -23,22 +23,25 @@ import org.apache.nifi.components.PropertyDescriptor; import org.apache.nifi.components.ValidationContext; import org.apache.nifi.controller.ValidationContextFactory; import org.apache.nifi.controller.service.ControllerServiceProvider; +import org.apache.nifi.registry.VariableRegistry; public class StandardValidationContextFactory implements ValidationContextFactory { private final ControllerServiceProvider serviceProvider; + private final VariableRegistry variableRegistry; - public StandardValidationContextFactory(final ControllerServiceProvider serviceProvider) { + public StandardValidationContextFactory(final ControllerServiceProvider serviceProvider, final VariableRegistry variableRegistry) { this.serviceProvider = serviceProvider; + this.variableRegistry = variableRegistry; } @Override public ValidationContext newValidationContext(final Map<PropertyDescriptor, String> properties, final String annotationData) { - return new StandardValidationContext(serviceProvider, properties, annotationData); + return new StandardValidationContext(serviceProvider, properties, annotationData, variableRegistry); } @Override public ValidationContext newValidationContext(final Set<String> serviceIdentifiersToNotValidate, final Map<PropertyDescriptor, String> properties, final String annotationData) { - return new StandardValidationContext(serviceProvider, serviceIdentifiersToNotValidate, properties, annotationData); + return new StandardValidationContext(serviceProvider, serviceIdentifiersToNotValidate, properties, annotationData, variableRegistry); } } http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java index 3b33478..c2d5cac 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java @@ -59,6 +59,8 @@ import org.apache.nifi.processor.ProcessSession; import org.apache.nifi.processor.Processor; import org.apache.nifi.processor.StandardValidationContextFactory; import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.registry.VariableRegistry; +import org.apache.nifi.registry.VariableRegistryUtils; import org.apache.nifi.reporting.AbstractReportingTask; import org.apache.nifi.reporting.InitializationException; import org.apache.nifi.reporting.ReportingContext; @@ -74,12 +76,13 @@ public class TestStandardProcessScheduler { private ReportingTaskNode taskNode = null; private TestReportingTask reportingTask = null; private StateManagerProvider stateMgrProvider = Mockito.mock(StateManagerProvider.class); + private VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry(); @Before public void setup() throws InitializationException { System.setProperty("nifi.properties.file.path", "src/test/resources/nifi.properties"); this.refreshNiFiProperties(); - scheduler = new StandardProcessScheduler(Mockito.mock(Heartbeater.class), Mockito.mock(ControllerServiceProvider.class), null, stateMgrProvider); + scheduler = new StandardProcessScheduler(Mockito.mock(Heartbeater.class), Mockito.mock(ControllerServiceProvider.class), null, stateMgrProvider, variableRegistry); scheduler.setSchedulingAgent(SchedulingStrategy.TIMER_DRIVEN, Mockito.mock(SchedulingAgent.class)); reportingTask = new TestReportingTask(); @@ -87,8 +90,8 @@ public class TestStandardProcessScheduler { Mockito.mock(ComponentLog.class), null); reportingTask.initialize(config); - final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(null); - taskNode = new StandardReportingTaskNode(reportingTask, UUID.randomUUID().toString(), null, scheduler, validationContextFactory); + final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(null, variableRegistry); + taskNode = new StandardReportingTaskNode(reportingTask, UUID.randomUUID().toString(), null, scheduler, validationContextFactory, variableRegistry); } /** @@ -119,10 +122,10 @@ public class TestStandardProcessScheduler { public void testDisableControllerServiceWithProcessorTryingToStartUsingIt() throws InterruptedException { final Processor proc = new ServiceReferencingProcessor(); - final ControllerServiceProvider serviceProvider = new StandardControllerServiceProvider(scheduler, null, Mockito.mock(StateManagerProvider.class)); + final ControllerServiceProvider serviceProvider = new StandardControllerServiceProvider(scheduler, null, Mockito.mock(StateManagerProvider.class), variableRegistry); final ControllerServiceNode service = serviceProvider.createControllerService(NoStartServiceImpl.class.getName(), "service", true); final ProcessorNode procNode = new StandardProcessorNode(proc, UUID.randomUUID().toString(), - new StandardValidationContextFactory(serviceProvider), scheduler, serviceProvider); + new StandardValidationContextFactory(serviceProvider, VariableRegistryUtils.createVariableRegistry()), scheduler, serviceProvider); procNode.setProperty(ServiceReferencingProcessor.SERVICE_DESC.getName(), service.getIdentifier()); @@ -200,7 +203,7 @@ public class TestStandardProcessScheduler { @Test public void validateServiceEnablementLogicHappensOnlyOnce() throws Exception { final ProcessScheduler scheduler = createScheduler(); - StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider); + StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider, VariableRegistryUtils.createVariableRegistry()); final ControllerServiceNode serviceNode = provider.createControllerService(SimpleTestService.class.getName(), "1", false); assertFalse(serviceNode.isActive()); @@ -239,7 +242,7 @@ public class TestStandardProcessScheduler { @Test public void validateDisabledServiceCantBeDisabled() throws Exception { final ProcessScheduler scheduler = createScheduler(); - StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider); + StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider, VariableRegistryUtils.createVariableRegistry()); final ControllerServiceNode serviceNode = provider.createControllerService(SimpleTestService.class.getName(), "1", false); SimpleTestService ts = (SimpleTestService) serviceNode.getControllerServiceImplementation(); @@ -277,7 +280,7 @@ public class TestStandardProcessScheduler { @Test public void validateEnabledServiceCanOnlyBeDisabledOnce() throws Exception { final ProcessScheduler scheduler = createScheduler(); - StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider); + StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider, VariableRegistryUtils.createVariableRegistry()); final ControllerServiceNode serviceNode = provider.createControllerService(SimpleTestService.class.getName(), "1", false); SimpleTestService ts = (SimpleTestService) serviceNode.getControllerServiceImplementation(); @@ -311,7 +314,7 @@ public class TestStandardProcessScheduler { @Test public void validateDisablingOfTheFailedService() throws Exception { final ProcessScheduler scheduler = createScheduler(); - StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider); + StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider, VariableRegistryUtils.createVariableRegistry()); final ControllerServiceNode serviceNode = provider.createControllerService(FailingService.class.getName(), "1", false); scheduler.enableControllerService(serviceNode); @@ -342,7 +345,7 @@ public class TestStandardProcessScheduler { @Test public void validateEnabledDisableMultiThread() throws Exception { final ProcessScheduler scheduler = createScheduler(); - StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider); + StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider, VariableRegistryUtils.createVariableRegistry()); ExecutorService executor = Executors.newCachedThreadPool(); for (int i = 0; i < 200; i++) { final ControllerServiceNode serviceNode = provider @@ -385,7 +388,7 @@ public class TestStandardProcessScheduler { @Test public void validateNeverEnablingServiceCanStillBeDisabled() throws Exception { final ProcessScheduler scheduler = createScheduler(); - StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider); + StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider, VariableRegistryUtils.createVariableRegistry()); final ControllerServiceNode serviceNode = provider.createControllerService(LongEnablingService.class.getName(), "1", false); LongEnablingService ts = (LongEnablingService) serviceNode.getControllerServiceImplementation(); @@ -410,7 +413,7 @@ public class TestStandardProcessScheduler { @Test public void validateLongEnablingServiceCanStillBeDisabled() throws Exception { final ProcessScheduler scheduler = createScheduler(); - StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider); + StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider, VariableRegistryUtils.createVariableRegistry()); final ControllerServiceNode serviceNode = provider.createControllerService(LongEnablingService.class.getName(), "1", false); LongEnablingService ts = (LongEnablingService) serviceNode.getControllerServiceImplementation(); @@ -507,6 +510,6 @@ public class TestStandardProcessScheduler { } private ProcessScheduler createScheduler() { - return new StandardProcessScheduler(mock(Heartbeater.class), null, null, stateMgrProvider); + return new StandardProcessScheduler(mock(Heartbeater.class), null, null, stateMgrProvider, VariableRegistryUtils.createVariableRegistry()); } } http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java index b9c0f7f..2390c79 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java @@ -22,6 +22,9 @@ import org.apache.nifi.controller.ControllerService; import org.apache.nifi.controller.StandardFlowServiceTest; import org.apache.nifi.nar.ExtensionManager; import org.apache.nifi.nar.NarClassLoaders; +import org.apache.nifi.registry.VariableRegistry; +import org.apache.nifi.registry.VariableRegistryFactory; +import org.apache.nifi.registry.VariableRegistryUtils; import org.apache.nifi.reporting.InitializationException; import org.apache.nifi.util.NiFiProperties; import org.junit.Before; @@ -33,6 +36,7 @@ public class StandardControllerServiceProviderTest { private ControllerService proxied; private ControllerService implementation; + private static VariableRegistry variableRegistry; @BeforeClass public static void setupSuite() throws Exception { @@ -40,6 +44,10 @@ public class StandardControllerServiceProviderTest { NiFiProperties properties = NiFiProperties.getInstance(); NarClassLoaders.load(properties); ExtensionManager.discoverExtensions(); + VariableRegistry propRegistry = VariableRegistryFactory.getPropertiesInstance(properties.getVariableRegistryPropertiesPaths()); + variableRegistry = VariableRegistryUtils.createVariableRegistry(); + variableRegistry.addRegistry(propRegistry); + } @Before @@ -67,7 +75,7 @@ public class StandardControllerServiceProviderTest { @Override public void onComponentRemoved(String componentId) { } - }); + }, variableRegistry); ControllerServiceNode node = provider.createControllerService(clazz, id, true); proxied = node.getProxiedControllerService(); implementation = node.getControllerServiceImplementation(); http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java index 5abefda..a11610d 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java @@ -41,6 +41,8 @@ import org.apache.nifi.controller.service.mock.ServiceB; import org.apache.nifi.groups.ProcessGroup; import org.apache.nifi.groups.StandardProcessGroup; import org.apache.nifi.processor.StandardValidationContextFactory; +import org.apache.nifi.registry.VariableRegistry; +import org.apache.nifi.registry.VariableRegistryUtils; import org.junit.Assert; import org.junit.BeforeClass; import org.junit.Test; @@ -70,6 +72,8 @@ public class TestStandardControllerServiceProvider { } }; + private static VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry(); + @BeforeClass public static void setNiFiProps() { System.setProperty("nifi.properties.file.path", "src/test/resources/nifi.properties"); @@ -77,13 +81,13 @@ public class TestStandardControllerServiceProvider { private StandardProcessScheduler createScheduler() { final Heartbeater heartbeater = Mockito.mock(Heartbeater.class); - return new StandardProcessScheduler(heartbeater, null, null, stateManagerProvider); + return new StandardProcessScheduler(heartbeater, null, null, stateManagerProvider, variableRegistry); } @Test public void testDisableControllerService() { final ProcessScheduler scheduler = createScheduler(); - final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider); + final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider, variableRegistry); final ControllerServiceNode serviceNode = provider.createControllerService(ServiceB.class.getName(), "B", false); provider.enableControllerService(serviceNode); @@ -93,7 +97,7 @@ public class TestStandardControllerServiceProvider { @Test(timeout=10000) public void testEnableDisableWithReference() { final ProcessScheduler scheduler = createScheduler(); - final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider); + final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider, variableRegistry); final ControllerServiceNode serviceNodeB = provider.createControllerService(ServiceB.class.getName(), "B", false); final ControllerServiceNode serviceNodeA = provider.createControllerService(ServiceA.class.getName(), "A", false); @@ -146,7 +150,7 @@ public class TestStandardControllerServiceProvider { } public void testEnableReferencingServicesGraph(ProcessScheduler scheduler) { - final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider); + final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider, variableRegistry); // build a graph of controller services with dependencies as such: // @@ -189,7 +193,7 @@ public class TestStandardControllerServiceProvider { @Test public void testOrderingOfServices() { - final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(null, null, stateManagerProvider); + final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(null, null, stateManagerProvider, variableRegistry); final ControllerServiceNode serviceNode1 = provider.createControllerService(ServiceA.class.getName(), "1", false); final ControllerServiceNode serviceNode2 = provider.createControllerService(ServiceB.class.getName(), "2", false); @@ -333,9 +337,9 @@ public class TestStandardControllerServiceProvider { private ProcessorNode createProcessor(final StandardProcessScheduler scheduler, final ControllerServiceProvider serviceProvider) { final ProcessorNode procNode = new StandardProcessorNode(new DummyProcessor(), UUID.randomUUID().toString(), - new StandardValidationContextFactory(serviceProvider), scheduler, serviceProvider); + new StandardValidationContextFactory(serviceProvider, null), scheduler, serviceProvider); - final ProcessGroup group = new StandardProcessGroup(UUID.randomUUID().toString(), serviceProvider, scheduler, null, null, null); + final ProcessGroup group = new StandardProcessGroup(UUID.randomUUID().toString(), serviceProvider, scheduler, null, null, null, variableRegistry); group.addProcessor(procNode); procNode.setProcessGroup(group); @@ -345,7 +349,7 @@ public class TestStandardControllerServiceProvider { @Test public void testEnableReferencingComponents() { final StandardProcessScheduler scheduler = createScheduler(); - final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(null, null, stateManagerProvider); + final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(null, null, stateManagerProvider, variableRegistry); final ControllerServiceNode serviceNode = provider.createControllerService(ServiceA.class.getName(), "1", false); final ProcessorNode procNode = createProcessor(scheduler, provider); http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/local/TestWriteAheadLocalStateProvider.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/local/TestWriteAheadLocalStateProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/local/TestWriteAheadLocalStateProvider.java index 9f7c4c9..d24441a 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/local/TestWriteAheadLocalStateProvider.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/local/TestWriteAheadLocalStateProvider.java @@ -32,6 +32,8 @@ import org.apache.nifi.components.state.StateProvider; import org.apache.nifi.components.state.StateProviderInitializationContext; import org.apache.nifi.controller.state.StateMapUpdate; import org.apache.nifi.controller.state.providers.AbstractTestStateProvider; +import org.apache.nifi.registry.VariableRegistry; +import org.apache.nifi.registry.VariableRegistryUtils; import org.junit.After; import org.junit.Before; import org.wali.WriteAheadRepository; @@ -43,9 +45,9 @@ public class TestWriteAheadLocalStateProvider extends AbstractTestStateProvider @Before public void setup() throws IOException { provider = new WriteAheadLocalStateProvider(); - + final VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry(); final Map<PropertyDescriptor, PropertyValue> properties = new HashMap<>(); - properties.put(WriteAheadLocalStateProvider.PATH, new StandardPropertyValue("target/local-state-provider/" + UUID.randomUUID().toString(), null)); + properties.put(WriteAheadLocalStateProvider.PATH, new StandardPropertyValue("target/local-state-provider/" + UUID.randomUUID().toString(), null, variableRegistry)); provider.initialize(new StateProviderInitializationContext() { @Override @@ -62,7 +64,7 @@ public class TestWriteAheadLocalStateProvider extends AbstractTestStateProvider public PropertyValue getProperty(final PropertyDescriptor property) { final PropertyValue prop = properties.get(property); if (prop == null) { - return new StandardPropertyValue(null, null); + return new StandardPropertyValue(null, null, variableRegistry); } return prop; } http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/zookeeper/TestZooKeeperStateProvider.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/zookeeper/TestZooKeeperStateProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/zookeeper/TestZooKeeperStateProvider.java index 3542a09..dd7c871 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/zookeeper/TestZooKeeperStateProvider.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/zookeeper/TestZooKeeperStateProvider.java @@ -36,6 +36,8 @@ import org.apache.nifi.components.state.StateProvider; import org.apache.nifi.components.state.StateProviderInitializationContext; import org.apache.nifi.components.state.exception.StateTooLargeException; import org.apache.nifi.controller.state.providers.AbstractTestStateProvider; +import org.apache.nifi.registry.VariableRegistry; +import org.apache.nifi.registry.VariableRegistryUtils; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.KeeperException.Code; import org.apache.zookeeper.ZooDefs.Perms; @@ -50,6 +52,7 @@ public class TestZooKeeperStateProvider extends AbstractTestStateProvider { private StateProvider provider; private TestingServer zkServer; + private VariableRegistry variableRegistry; private static final Map<PropertyDescriptor, String> defaultProperties = new HashMap<>(); @@ -68,6 +71,7 @@ public class TestZooKeeperStateProvider extends AbstractTestStateProvider { final Map<PropertyDescriptor, String> properties = new HashMap<>(defaultProperties); properties.put(ZooKeeperStateProvider.CONNECTION_STRING, zkServer.getConnectString()); this.provider = createProvider(properties); + variableRegistry = VariableRegistryUtils.createVariableRegistry(); } private void initializeProvider(final ZooKeeperStateProvider provider, final Map<PropertyDescriptor, String> properties) throws IOException { @@ -81,7 +85,7 @@ public class TestZooKeeperStateProvider extends AbstractTestStateProvider { public Map<PropertyDescriptor, PropertyValue> getProperties() { final Map<PropertyDescriptor, PropertyValue> propValueMap = new HashMap<>(); for (final Map.Entry<PropertyDescriptor, String> entry : properties.entrySet()) { - propValueMap.put(entry.getKey(), new StandardPropertyValue(entry.getValue(), null)); + propValueMap.put(entry.getKey(), new StandardPropertyValue(entry.getValue(), null, variableRegistry)); } return propValueMap; } @@ -89,7 +93,7 @@ public class TestZooKeeperStateProvider extends AbstractTestStateProvider { @Override public PropertyValue getProperty(final PropertyDescriptor property) { final String prop = properties.get(property); - return new StandardPropertyValue(prop, null); + return new StandardPropertyValue(prop, null, variableRegistry); } @Override
