NIFI-2033: Allow Controller Services to be scoped at Controller level instead of just group level. This closes #540
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/c955ec16 Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/c955ec16 Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/c955ec16 Branch: refs/heads/master Commit: c955ec1689da0bfec9531a6698acc67780a55788 Parents: 2c69c25 Author: Mark Payne <[email protected]> Authored: Fri Jun 17 09:08:44 2016 -0400 Committer: Matt Gilman <[email protected]> Committed: Fri Jun 17 13:26:30 2016 -0400 ---------------------------------------------------------------------- .../nifi/components/PropertyDescriptor.java | 6 +- .../controller/ControllerServiceLookup.java | 3 +- .../nifi/web/NiFiWebConfigurationContext.java | 3 +- .../nifi/util/MockControllerServiceLookup.java | 6 +- .../MockProcessorInitializationContext.java | 12 +- .../apache/nifi/util/MockValidationContext.java | 6 +- .../mock/MockControllerServiceLookup.java | 14 +- .../controller/AbstractConfiguredComponent.java | 4 +- .../controller/ValidationContextFactory.java | 5 +- .../service/ControllerServiceProvider.java | 22 + .../apache/nifi/controller/FlowController.java | 73 +++- .../controller/StandardFlowSynchronizer.java | 4 +- .../nifi/controller/StandardProcessorNode.java | 28 +- .../reporting/StandardReportingContext.java | 4 +- .../StandardReportingInitializationContext.java | 4 +- .../serialization/StandardFlowSerializer.java | 6 +- .../service/ControllerServiceLoader.java | 15 +- ...dControllerServiceInitializationContext.java | 6 +- .../StandardControllerServiceProvider.java | 110 +++-- .../manager/StandardStateManagerProvider.java | 2 +- ...omponentSpecificControllerServiceLookup.java | 66 +++ .../nifi/processor/StandardProcessContext.java | 10 +- .../processor/StandardValidationContext.java | 17 +- .../StandardValidationContextFactory.java | 8 +- .../TestStandardProcessScheduler.java | 81 ++-- .../StandardControllerServiceProviderTest.java | 2 +- .../TestStandardControllerServiceProvider.java | 41 +- .../service/mock/MockProcessGroup.java | 169 ++++---- .../processor/TestStandardPropertyValue.java | 10 +- .../nifi/web/StandardNiFiServiceFacade.java | 404 ++++++++++--------- .../StandardNiFiWebConfigurationContext.java | 114 +++--- .../org/apache/nifi/web/api/dto/DtoFactory.java | 51 +-- .../nifi/web/dao/ControllerServiceDAO.java | 5 +- .../dao/impl/StandardControllerServiceDAO.java | 66 ++- .../src/main/resources/nifi-web-api-context.xml | 3 +- 35 files changed, 826 insertions(+), 554 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/nifi-api/src/main/java/org/apache/nifi/components/PropertyDescriptor.java ---------------------------------------------------------------------- diff --git a/nifi-api/src/main/java/org/apache/nifi/components/PropertyDescriptor.java b/nifi-api/src/main/java/org/apache/nifi/components/PropertyDescriptor.java index a8b9bdf..7b382f6 100644 --- a/nifi-api/src/main/java/org/apache/nifi/components/PropertyDescriptor.java +++ b/nifi-api/src/main/java/org/apache/nifi/components/PropertyDescriptor.java @@ -138,7 +138,7 @@ public final class PropertyDescriptor implements Comparable<PropertyDescriptor> // if the property descriptor identifies a Controller Service, validate that the ControllerService exists, is of the correct type, and is valid if (controllerServiceDefinition != null) { - final Set<String> validIdentifiers = context.getControllerServiceLookup().getControllerServiceIdentifiers(controllerServiceDefinition, context.getProcessGroupIdentifier()); + final Set<String> validIdentifiers = context.getControllerServiceLookup().getControllerServiceIdentifiers(controllerServiceDefinition); if (validIdentifiers != null && validIdentifiers.contains(input)) { final ControllerService controllerService = context.getControllerServiceLookup().getControllerService(input); if (!context.isValidationRequired(controllerService)) { @@ -213,7 +213,7 @@ public final class PropertyDescriptor implements Comparable<PropertyDescriptor> * for ENABLED state even though by the time this method returns the * dependent service's state could be fully ENABLED. */ - private boolean isDependentServiceEnableable(ValidationContext context, String serviceId) { + private boolean isDependentServiceEnableable(final ValidationContext context, final String serviceId) { boolean enableable = context.getControllerServiceLookup().isControllerServiceEnabling(serviceId); if (!enableable) { enableable = context.getControllerServiceLookup().isControllerServiceEnabled(serviceId); @@ -516,7 +516,7 @@ public final class PropertyDescriptor implements Comparable<PropertyDescriptor> return true; } - PropertyDescriptor desc = (PropertyDescriptor) other; + final PropertyDescriptor desc = (PropertyDescriptor) other; return this.name.equals(desc.name); } http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/nifi-api/src/main/java/org/apache/nifi/controller/ControllerServiceLookup.java ---------------------------------------------------------------------- diff --git a/nifi-api/src/main/java/org/apache/nifi/controller/ControllerServiceLookup.java b/nifi-api/src/main/java/org/apache/nifi/controller/ControllerServiceLookup.java index 3345a92..03f6e9b 100644 --- a/nifi-api/src/main/java/org/apache/nifi/controller/ControllerServiceLookup.java +++ b/nifi-api/src/main/java/org/apache/nifi/controller/ControllerServiceLookup.java @@ -56,13 +56,12 @@ public interface ControllerServiceLookup { /** * * @param serviceType type of service to get identifiers for - * @param groupId the ID of the Process Group to look in for Controller Services * * @return the set of all Controller Service Identifiers whose Controller * Service is of the given type. * @throws IllegalArgumentException if the given class is not an interface */ - Set<String> getControllerServiceIdentifiers(Class<? extends ControllerService> serviceType, String groupId) throws IllegalArgumentException; + Set<String> getControllerServiceIdentifiers(Class<? extends ControllerService> serviceType) throws IllegalArgumentException; /** * @param serviceIdentifier identifier to look up http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/nifi-api/src/main/java/org/apache/nifi/web/NiFiWebConfigurationContext.java ---------------------------------------------------------------------- diff --git a/nifi-api/src/main/java/org/apache/nifi/web/NiFiWebConfigurationContext.java b/nifi-api/src/main/java/org/apache/nifi/web/NiFiWebConfigurationContext.java index 39bea4f..e2689af 100644 --- a/nifi-api/src/main/java/org/apache/nifi/web/NiFiWebConfigurationContext.java +++ b/nifi-api/src/main/java/org/apache/nifi/web/NiFiWebConfigurationContext.java @@ -28,12 +28,13 @@ public interface NiFiWebConfigurationContext { /** * @param serviceIdentifier of the controller service + * @param componentId the id of the component that is referencing the controller service * @return the ControllerService for the specified identifier. If a * corresponding service cannot be found, null is returned. If this NiFi is * clustered, the only services available will be those those availability * is NCM only */ - ControllerService getControllerService(String serviceIdentifier); + ControllerService getControllerService(String serviceIdentifier, String componentId); /** * Provides a mechanism for custom UIs to save actions to appear in NiFi http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/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 d6ff5c8..e07540b 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 @@ -38,7 +38,7 @@ public abstract class MockControllerServiceLookup implements ControllerServiceLo return addControllerService(service, service.getIdentifier()); } - public void removeControllerService(ControllerService service) { + public void removeControllerService(final ControllerService service) { final ControllerService canonical = getControllerService(service.getIdentifier()); if (canonical == null || canonical != service) { throw new IllegalArgumentException("Controller Service " + service + " is not known"); @@ -82,7 +82,7 @@ public abstract class MockControllerServiceLookup implements ControllerServiceLo } @Override - public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType, String groupId) { + public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType) { final Set<String> ids = new HashSet<>(); for (final Map.Entry<String, ControllerServiceConfiguration> entry : controllerServiceMap.entrySet()) { if (serviceType.isAssignableFrom(entry.getValue().getService().getClass())) { @@ -93,7 +93,7 @@ public abstract class MockControllerServiceLookup implements ControllerServiceLo } @Override - public String getControllerServiceName(String serviceIdentifier) { + public String getControllerServiceName(final String serviceIdentifier) { final ControllerServiceConfiguration status = controllerServiceMap.get(serviceIdentifier); return status == null ? null : serviceIdentifier; } http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/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 f0961ef..82295f9 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 @@ -47,8 +47,8 @@ public class MockProcessorInitializationContext implements ProcessorInitializati } @Override - public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType, String groupId) { - return context.getControllerServiceIdentifiers(serviceType, groupId); + public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType) { + return context.getControllerServiceIdentifiers(serviceType); } @Override @@ -62,22 +62,22 @@ public class MockProcessorInitializationContext implements ProcessorInitializati } @Override - public String getControllerServiceName(String serviceIdentifier) { + public String getControllerServiceName(final String serviceIdentifier) { return context.getControllerServiceName(serviceIdentifier); } @Override - public boolean isControllerServiceEnabled(String serviceIdentifier) { + public boolean isControllerServiceEnabled(final String serviceIdentifier) { return context.isControllerServiceEnabled(serviceIdentifier); } @Override - public boolean isControllerServiceEnabled(ControllerService service) { + public boolean isControllerServiceEnabled(final ControllerService service) { return context.isControllerServiceEnabled(service); } @Override - public boolean isControllerServiceEnabling(String serviceIdentifier) { + public boolean isControllerServiceEnabling(final String serviceIdentifier) { return context.isControllerServiceEnabling(serviceIdentifier); } } http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/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 84d3277..f0ff58b 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 @@ -86,8 +86,8 @@ public class MockValidationContext implements ValidationContext, ControllerServi } @Override - public Set<String> getControllerServiceIdentifiers(Class<? extends ControllerService> serviceType, String groupId) { - return context.getControllerServiceIdentifiers(serviceType, groupId); + public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType) { + return context.getControllerServiceIdentifiers(serviceType); } @Override @@ -117,7 +117,7 @@ public class MockValidationContext implements ValidationContext, ControllerServi } @Override - public boolean isControllerServiceEnabling(String serviceIdentifier) { + public boolean isControllerServiceEnabling(final String serviceIdentifier) { return context.isControllerServiceEnabling(serviceIdentifier); } http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-documentation/src/main/java/org/apache/nifi/documentation/mock/MockControllerServiceLookup.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-documentation/src/main/java/org/apache/nifi/documentation/mock/MockControllerServiceLookup.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-documentation/src/main/java/org/apache/nifi/documentation/mock/MockControllerServiceLookup.java index 26a6f39..e92e801 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-documentation/src/main/java/org/apache/nifi/documentation/mock/MockControllerServiceLookup.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-documentation/src/main/java/org/apache/nifi/documentation/mock/MockControllerServiceLookup.java @@ -32,34 +32,32 @@ import org.apache.nifi.controller.ControllerServiceLookup; public class MockControllerServiceLookup implements ControllerServiceLookup { @Override - public ControllerService getControllerService(String serviceIdentifier) { + public ControllerService getControllerService(final String serviceIdentifier) { return null; } @Override - public boolean isControllerServiceEnabled(String serviceIdentifier) { + public boolean isControllerServiceEnabled(final String serviceIdentifier) { return false; } @Override - public boolean isControllerServiceEnabled(ControllerService service) { + public boolean isControllerServiceEnabled(final ControllerService service) { return false; } @Override - public Set<String> getControllerServiceIdentifiers(Class<? extends ControllerService> serviceType, String groupId) - throws IllegalArgumentException { + public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType) throws IllegalArgumentException { return Collections.emptySet(); } @Override - public boolean isControllerServiceEnabling(String serviceIdentifier) { + public boolean isControllerServiceEnabling(final String serviceIdentifier) { return false; } @Override - public String getControllerServiceName(String serviceIdentifier) { + public String getControllerServiceName(final String serviceIdentifier) { return serviceIdentifier; } - } http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/AbstractConfiguredComponent.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/AbstractConfiguredComponent.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/AbstractConfiguredComponent.java index b3de995..0454a2f 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/AbstractConfiguredComponent.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/AbstractConfiguredComponent.java @@ -262,7 +262,7 @@ public abstract class AbstractConfiguredComponent implements ConfigurableCompone @Override public boolean isValid() { final Collection<ValidationResult> validationResults = validate(validationContextFactory.newValidationContext( - getProperties(), getAnnotationData(), getProcessGroupIdentifier())); + getProperties(), getAnnotationData(), getProcessGroupIdentifier(), getIdentifier())); for (final ValidationResult result : validationResults) { if (!result.isValid()) { @@ -283,7 +283,7 @@ public abstract class AbstractConfiguredComponent implements ConfigurableCompone lock.lock(); try { final ValidationContext validationContext = validationContextFactory.newValidationContext( - serviceIdentifiersNotToValidate, getProperties(), getAnnotationData(), getProcessGroupIdentifier()); + serviceIdentifiersNotToValidate, getProperties(), getAnnotationData(), getProcessGroupIdentifier(), getIdentifier()); final Collection<ValidationResult> validationResults; try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ValidationContextFactory.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ValidationContextFactory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ValidationContextFactory.java index be6346f..1f17d39 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ValidationContextFactory.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ValidationContextFactory.java @@ -24,8 +24,9 @@ import org.apache.nifi.components.ValidationContext; public interface ValidationContextFactory { - ValidationContext newValidationContext(Map<PropertyDescriptor, String> properties, String annotationData, String groupId); + ValidationContext newValidationContext(Map<PropertyDescriptor, String> properties, String annotationData, String groupId, String componentId); - ValidationContext newValidationContext(Set<String> serviceIdentifiersToNotValidate, Map<PropertyDescriptor, String> properties, String annotationData, String groupId); + ValidationContext newValidationContext(Set<String> serviceIdentifiersToNotValidate, Map<PropertyDescriptor, String> properties, + String annotationData, String groupId, String componentId); } http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java index 4db6bc9..51d54a0 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java @@ -21,6 +21,7 @@ import java.util.Set; import org.apache.nifi.annotation.lifecycle.OnAdded; import org.apache.nifi.controller.ConfiguredComponent; +import org.apache.nifi.controller.ControllerService; import org.apache.nifi.controller.ControllerServiceLookup; /** @@ -176,4 +177,25 @@ public interface ControllerServiceProvider extends ControllerServiceLookup { * @param serviceNode the node */ Set<ConfiguredComponent> scheduleReferencingComponents(ControllerServiceNode serviceNode); + + /** + * + * @param serviceType type of service to get identifiers for + * @param groupId the ID of the Process Group to look in for Controller Services + * + * @return the set of all Controller Service Identifiers whose Controller + * Service is of the given type. + * @throws IllegalArgumentException if the given class is not an interface + */ + Set<String> getControllerServiceIdentifiers(Class<? extends ControllerService> serviceType, String groupId) throws IllegalArgumentException; + + /** + * @param serviceIdentifier the identifier of the controller service + * @param componentId the identifier of the component that is referencing the service. + * @return the Controller Service that is registered with the given identifier or <code>null</code> if that + * identifier does not exist for any controller service or if the controller service with that identifier is + * not accessible from the component with the given componentId, or if no component exists with the given + * identifier + */ + ControllerService getControllerServiceForComponent(String serviceIdentifier, String componentId); } http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/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 f4e70af..cf2cba6 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 @@ -278,6 +278,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 ConcurrentMap<String, ControllerServiceNode> rootControllerServices = new ConcurrentHashMap<>(); private volatile ZooKeeperStateServer zooKeeperStateServer; @@ -501,8 +502,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R rootGroup.setName(DEFAULT_ROOT_GROUP_NAME); instanceId = UUID.randomUUID().toString(); - controllerServiceProvider = new StandardControllerServiceProvider(processScheduler, bulletinRepository, stateManagerProvider); - controllerServiceProvider.setRootProcessGroup(rootGroup); + controllerServiceProvider = new StandardControllerServiceProvider(this, processScheduler, bulletinRepository, stateManagerProvider); if (remoteInputSocketPort == null) { LOG.info("Not enabling RAW Socket Site-to-Site functionality because nifi.remote.input.socket.port is not set"); @@ -521,7 +521,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R externalSiteListeners.add(HttpRemoteSiteListener.getInstance()); } - for(RemoteSiteListener listener : externalSiteListeners) { + for(final RemoteSiteListener listener : externalSiteListeners) { listener.setRootGroup(rootGroup); } @@ -659,7 +659,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R // ContentRepository to purge superfluous files contentRepository.cleanup(); - for(RemoteSiteListener listener : externalSiteListeners) { + for(final RemoteSiteListener listener : externalSiteListeners) { listener.start(); } @@ -1297,7 +1297,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R + "will take an indeterminate amount of time to stop. Might need to kill the program manually."); } - for(RemoteSiteListener listener : externalSiteListeners) { + for(final RemoteSiteListener listener : externalSiteListeners) { listener.stop(); } @@ -1442,12 +1442,10 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R try { rootGroup = group; - for(RemoteSiteListener listener : externalSiteListeners) { + for(final RemoteSiteListener listener : externalSiteListeners) { listener.setRootGroup(rootGroup); } - controllerServiceProvider.setRootProcessGroup(rootGroup); - // update the heartbeat bean this.heartbeatBeanRef.set(new HeartbeatBean(rootGroup, isPrimary(), connectionStatus)); } finally { @@ -2875,10 +2873,67 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R } @Override + public ControllerService getControllerServiceForComponent(final String serviceIdentifier, final String componentId) { + return controllerServiceProvider.getControllerServiceForComponent(serviceIdentifier, componentId); + } + + @Override + public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType) throws IllegalArgumentException { + return controllerServiceProvider.getControllerServiceIdentifiers(serviceType); + } + + @Override public ControllerServiceNode getControllerServiceNode(final String serviceIdentifier) { return controllerServiceProvider.getControllerServiceNode(serviceIdentifier); } + public Set<ControllerServiceNode> getRootControllerServices() { + return new HashSet<>(rootControllerServices.values()); + } + + public void addRootControllerService(final ControllerServiceNode serviceNode) { + final ControllerServiceNode existing = rootControllerServices.putIfAbsent(serviceNode.getIdentifier(), serviceNode); + if (existing != null) { + throw new IllegalStateException("Controller Service with ID " + serviceNode.getIdentifier() + " already exists at the Controller level"); + } + } + + public ControllerServiceNode getRootControllerService(final String serviceIdentifier) { + return rootControllerServices.get(serviceIdentifier); + } + + public void removeRootControllerService(final ControllerServiceNode service) { + final ControllerServiceNode existing = rootControllerServices.get(requireNonNull(service).getIdentifier()); + if (existing == null) { + throw new IllegalStateException(service + " is not a member of this Process Group"); + } + + service.verifyCanDelete(); + + try (final NarCloseable x = NarCloseable.withNarLoader()) { + final ConfigurationContext configurationContext = new StandardConfigurationContext(service, controllerServiceProvider, null); + ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnRemoved.class, service.getControllerServiceImplementation(), configurationContext); + } + + for (final Map.Entry<PropertyDescriptor, String> entry : service.getProperties().entrySet()) { + final PropertyDescriptor descriptor = entry.getKey(); + if (descriptor.getControllerServiceDefinition() != null) { + final String value = entry.getValue() == null ? descriptor.getDefaultValue() : entry.getValue(); + if (value != null) { + final ControllerServiceNode referencedNode = getRootControllerService(value); + if (referencedNode != null) { + referencedNode.removeReference(service); + } + } + } + } + + rootControllerServices.remove(service.getIdentifier()); + getStateManagerProvider().onComponentRemoved(service.getIdentifier()); + + LOG.info("{} removed from Flow Controller", service, this); + } + @Override public boolean isControllerServiceEnabled(final ControllerService service) { return controllerServiceProvider.isControllerServiceEnabled(service); @@ -3844,7 +3899,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R } @Override - public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType, String groupId) { + public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType, final String groupId) { return controllerServiceProvider.getControllerServiceIdentifiers(serviceType, groupId); } http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java index 6b938d6..572fd79 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java @@ -314,7 +314,7 @@ public class StandardFlowSynchronizer implements FlowSynchronizer { } } - void scaleRootGroup(ProcessGroup rootGroup, FlowEncodingVersion encodingVersion) { + void scaleRootGroup(final ProcessGroup rootGroup, final FlowEncodingVersion encodingVersion) { if (encodingVersion == null || encodingVersion.getMajorVersion() < 1) { // Calculate new Positions if the encoding version of the flow is older than 1.0. PositionScaler.scale(rootGroup, 1.5, 1.34); @@ -939,7 +939,7 @@ public class StandardFlowSynchronizer implements FlowSynchronizer { remoteGroup.setYieldDuration(remoteGroupDto.getYieldDuration()); } - String transportProtocol = remoteGroupDto.getTransportProtocol(); + final String transportProtocol = remoteGroupDto.getTransportProtocol(); if (transportProtocol != null && !transportProtocol.trim().isEmpty()) { remoteGroup.setTransportProtocol(SiteToSiteTransportProtocol.valueOf(transportProtocol.toUpperCase())); } http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java index 1f10e6a..3ea2b6b 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java @@ -233,7 +233,7 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable } @Override - public void setPosition(Position position) { + public void setPosition(final Position position) { this.position.set(position); } @@ -373,7 +373,7 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable */ @SuppressWarnings("deprecation") public String getProcessorDescription() { - CapabilityDescription capDesc = processor.getClass().getAnnotation(CapabilityDescription.class); + final CapabilityDescription capDesc = processor.getClass().getAnnotation(CapabilityDescription.class); String description = null; if (capDesc != null) { description = capDesc.value(); @@ -644,7 +644,7 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable @Override public Set<Connection> getConnections(final Relationship relationship) { - Set<Connection> applicableConnections = connections.get(relationship); + final Set<Connection> applicableConnections = connections.get(relationship); return (applicableConnections == null) ? Collections.<Connection> emptySet() : Collections.unmodifiableSet(applicableConnections); } @@ -923,7 +923,7 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable public boolean isValid() { try { final ValidationContext validationContext = this.getValidationContextFactory() - .newValidationContext(getProperties(), getAnnotationData(), getProcessGroupIdentifier()); + .newValidationContext(getProperties(), getAnnotationData(), getProcessGroupIdentifier(), getIdentifier()); final Collection<ValidationResult> validationResults; try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { @@ -970,7 +970,7 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable final List<ValidationResult> results = new ArrayList<>(); try { final ValidationContext validationContext = this.getValidationContextFactory() - .newValidationContext(getProperties(), getAnnotationData(), getProcessGroup().getIdentifier()); + .newValidationContext(getProperties(), getAnnotationData(), getProcessGroup().getIdentifier(), getIdentifier()); final Collection<ValidationResult> validationResults; try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { @@ -1286,7 +1286,7 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable }; taskScheduler.execute(startProcRunnable); } else { - String procName = this.processor.getClass().getSimpleName(); + final String procName = this.processor.getClass().getSimpleName(); LOG.warn("Can not start '" + procName + "' since it's already in the process of being started or it is DISABLED - " + scheduledState.get()); @@ -1345,7 +1345,7 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable } else { scheduler.schedule(this, 100, TimeUnit.MILLISECONDS); } - } catch (Exception e) { + } catch (final Exception e) { LOG.warn("Failed while shutting down processor " + processor, e); } } @@ -1389,19 +1389,19 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable * be logged (WARN) informing a user so further actions could be taken. * </p> */ - private <T> void invokeTaskAsCancelableFuture(SchedulingAgentCallback callback, Callable<T> task) { - String timeoutString = NiFiProperties.getInstance().getProperty(NiFiProperties.PROCESSOR_SCHEDULING_TIMEOUT); - long onScheduleTimeout = timeoutString == null ? 60000 + private <T> void invokeTaskAsCancelableFuture(final SchedulingAgentCallback callback, final Callable<T> task) { + final String timeoutString = NiFiProperties.getInstance().getProperty(NiFiProperties.PROCESSOR_SCHEDULING_TIMEOUT); + final long onScheduleTimeout = timeoutString == null ? 60000 : FormatUtils.getTimeDuration(timeoutString.trim(), TimeUnit.MILLISECONDS); - Future<?> taskFuture = callback.invokeMonitoringTask(task); + final Future<?> taskFuture = callback.invokeMonitoringTask(task); try { taskFuture.get(onScheduleTimeout, TimeUnit.MILLISECONDS); - } catch (InterruptedException e) { + } catch (final InterruptedException e) { LOG.warn("Thread was interrupted while waiting for processor '" + this.processor.getClass().getSimpleName() + "' lifecycle OnScheduled operation to finish."); Thread.currentThread().interrupt(); throw new RuntimeException("Interrupted while executing one of processor's OnScheduled tasks.", e); - } catch (TimeoutException e) { + } catch (final TimeoutException e) { taskFuture.cancel(true); LOG.warn("Timed out while waiting for OnScheduled of '" + this.processor.getClass().getSimpleName() @@ -1411,7 +1411,7 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable + "eventually requiring NiFi to be restarted. This is usually a bug in the target Processor '" + this.processor + "' that needs to be documented, reported and eventually fixed."); throw new RuntimeException("Timed out while executing one of processor's OnScheduled task.", e); - } catch (ExecutionException e){ + } catch (final ExecutionException e){ throw new RuntimeException("Failed while executing one of processor's OnScheduled task.", e); } finally { callback.postMonitor(); http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/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 ff767ef..b174c4c 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 @@ -115,8 +115,8 @@ public class StandardReportingContext implements ReportingContext, ControllerSer } @Override - public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType, String groupId) { - return serviceProvider.getControllerServiceIdentifiers(serviceType, groupId); + public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType) { + return serviceProvider.getControllerServiceIdentifiers(serviceType, null); } @Override http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/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 8b7b3bf..caf38a8 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 @@ -75,8 +75,8 @@ public class StandardReportingInitializationContext implements ReportingInitiali } @Override - public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType, String groupId) { - return serviceProvider.getControllerServiceIdentifiers(serviceType, groupId); + public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType) { + return serviceProvider.getControllerServiceIdentifiers(serviceType, null); } @Override http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/StandardFlowSerializer.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/StandardFlowSerializer.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/StandardFlowSerializer.java index e774637..0892f41 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/StandardFlowSerializer.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/StandardFlowSerializer.java @@ -94,8 +94,12 @@ public class StandardFlowSerializer implements FlowSerializer { addTextElement(rootNode, "maxEventDrivenThreadCount", controller.getMaxEventDrivenThreadCount()); addProcessGroup(rootNode, controller.getGroup(controller.getRootGroupId()), "rootGroup"); + // Add root-level controller services final Element controllerServicesNode = doc.createElement("controllerServices"); rootNode.appendChild(controllerServicesNode); + for (final ControllerServiceNode serviceNode : controller.getRootControllerServices()) { + addControllerService(controllerServicesNode, serviceNode); + } final Element reportingTasksNode = doc.createElement("reportingTasks"); rootNode.appendChild(reportingTasksNode); @@ -252,7 +256,7 @@ public class StandardFlowSerializer implements FlowSerializer { } addTextElement(element, "proxyUser", remoteRef.getProxyUser()); if (!StringUtils.isEmpty(remoteRef.getProxyPassword())) { - String value = ENC_PREFIX + encryptor.encrypt(remoteRef.getProxyPassword()) + ENC_SUFFIX; + final String value = ENC_PREFIX + encryptor.encrypt(remoteRef.getProxyPassword()) + ENC_SUFFIX; addTextElement(element, "proxyPassword", value); } http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/ControllerServiceLoader.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/ControllerServiceLoader.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/ControllerServiceLoader.java index 74533ff..d302cff 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/ControllerServiceLoader.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/ControllerServiceLoader.java @@ -31,6 +31,7 @@ import javax.xml.parsers.DocumentBuilder; import javax.xml.parsers.DocumentBuilderFactory; import javax.xml.parsers.ParserConfigurationException; +import org.apache.nifi.controller.FlowController; import org.apache.nifi.controller.serialization.FlowFromDOMFactory; import org.apache.nifi.encrypt.StringEncryptor; import org.apache.nifi.groups.ProcessGroup; @@ -48,7 +49,7 @@ public class ControllerServiceLoader { private static final Logger logger = LoggerFactory.getLogger(ControllerServiceLoader.class); - public static List<ControllerServiceNode> loadControllerServices(final ControllerServiceProvider provider, final InputStream serializedStream, final ProcessGroup parentGroup, + public static List<ControllerServiceNode> loadControllerServices(final FlowController controller, final InputStream serializedStream, final ProcessGroup parentGroup, final StringEncryptor encryptor, final BulletinRepository bulletinRepo, final boolean autoResumeState) throws IOException { final DocumentBuilderFactory documentBuilderFactory = DocumentBuilderFactory.newInstance(); @@ -90,19 +91,21 @@ public class ControllerServiceLoader { final Document document = builder.parse(in); final Element controllerServices = document.getDocumentElement(); final List<Element> serviceElements = DomUtils.getChildElementsByTagName(controllerServices, "controllerService"); - return new ArrayList<>(loadControllerServices(serviceElements, provider, parentGroup, encryptor, bulletinRepo, autoResumeState)); + return new ArrayList<>(loadControllerServices(serviceElements, controller, parentGroup, encryptor, bulletinRepo, autoResumeState)); } catch (SAXException | ParserConfigurationException sxe) { throw new IOException(sxe); } } - public static Collection<ControllerServiceNode> loadControllerServices(final List<Element> serviceElements, final ControllerServiceProvider provider, final ProcessGroup parentGroup, + public static Collection<ControllerServiceNode> loadControllerServices(final List<Element> serviceElements, final FlowController controller, final ProcessGroup parentGroup, final StringEncryptor encryptor, final BulletinRepository bulletinRepo, final boolean autoResumeState) { final Map<ControllerServiceNode, Element> nodeMap = new HashMap<>(); for (final Element serviceElement : serviceElements) { - final ControllerServiceNode serviceNode = createControllerService(provider, serviceElement, encryptor); - if (parentGroup != null) { + final ControllerServiceNode serviceNode = createControllerService(controller, serviceElement, encryptor); + if (parentGroup == null) { + controller.addRootControllerService(serviceNode); + } else { parentGroup.addControllerService(serviceNode); } @@ -132,7 +135,7 @@ public class ControllerServiceLoader { } } - provider.enableControllerServices(nodesToEnable); + controller.enableControllerServices(nodesToEnable); } return nodeMap.keySet(); http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/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 fc6d376..71cd793 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 @@ -49,8 +49,8 @@ public class StandardControllerServiceInitializationContext implements Controlle } @Override - public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType, String groupId) { - return serviceProvider.getControllerServiceIdentifiers(serviceType, groupId); + public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType) { + return serviceProvider.getControllerServiceIdentifiers(serviceType); } @Override @@ -69,7 +69,7 @@ public class StandardControllerServiceInitializationContext implements Controlle } @Override - public boolean isControllerServiceEnabling(String serviceIdentifier) { + public boolean isControllerServiceEnabling(final String serviceIdentifier) { return serviceProvider.isControllerServiceEnabling(serviceIdentifier); } http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/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 425ee40..4e3249f 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 @@ -72,7 +72,7 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi private static final Set<Method> validDisabledMethods; private final BulletinRepository bulletinRepo; private final StateManagerProvider stateManagerProvider; - private volatile ProcessGroup rootGroup; + private final FlowController flowController; static { // methods that are okay to be called when the service is disabled. @@ -86,17 +86,15 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi validDisabledMethods = Collections.unmodifiableSet(validMethods); } - public StandardControllerServiceProvider(final ProcessScheduler scheduler, final BulletinRepository bulletinRepo, final StateManagerProvider stateManagerProvider) { - // 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 + public StandardControllerServiceProvider(final FlowController flowController, final ProcessScheduler scheduler, final BulletinRepository bulletinRepo, + final StateManagerProvider stateManagerProvider) { + + this.flowController = flowController; this.processScheduler = scheduler; this.bulletinRepo = bulletinRepo; this.stateManagerProvider = stateManagerProvider; } - public void setRootProcessGroup(ProcessGroup rootGroup) { - this.rootGroup = rootGroup; - } private Class<?>[] getInterfaces(final Class<?> cls) { final List<Class<?>> allIfcs = new ArrayList<>(); @@ -519,6 +517,52 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi return node == null ? null : node.getProxiedControllerService(); } + private ProcessGroup getRootGroup() { + return flowController.getGroup(flowController.getRootGroupId()); + } + + @Override + public ControllerService getControllerServiceForComponent(final String serviceIdentifier, final String componentId) { + final ProcessGroup rootGroup = getRootGroup(); + + // Find the Process Group that owns the component. + ProcessGroup groupOfInterest = null; + + final ProcessorNode procNode = rootGroup.findProcessor(componentId); + if (procNode == null) { + final ControllerServiceNode serviceNode = getControllerServiceNode(componentId); + if (serviceNode == null) { + final ReportingTaskNode taskNode = flowController.getReportingTaskNode(componentId); + if (taskNode == null) { + throw new IllegalStateException("Could not find any Processor, Reporting Task, or Controller Service with identifier " + componentId); + } + + // we have confirmed that the component is a reporting task. We can only reference Controller Services + // that are scoped at the FlowController level in this case. + final ControllerServiceNode rootServiceNode = flowController.getRootControllerService(serviceIdentifier); + return (rootServiceNode == null) ? null : rootServiceNode.getProxiedControllerService(); + } else { + groupOfInterest = serviceNode.getProcessGroup(); + } + } else { + groupOfInterest = procNode.getProcessGroup(); + } + + if (groupOfInterest == null) { + final ControllerServiceNode rootServiceNode = flowController.getRootControllerService(serviceIdentifier); + return (rootServiceNode == null) ? null : rootServiceNode.getProxiedControllerService(); + } + + final Set<ControllerServiceNode> servicesForGroup = groupOfInterest.getControllerServices(true); + for (final ControllerServiceNode serviceNode : servicesForGroup) { + if (serviceIdentifier.equals(serviceNode.getIdentifier())) { + return serviceNode.getProxiedControllerService(); + } + } + + return null; + } + @Override public boolean isControllerServiceEnabled(final ControllerService service) { return isControllerServiceEnabled(service.getIdentifier()); @@ -538,27 +582,33 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi @Override public ControllerServiceNode getControllerServiceNode(final String serviceIdentifier) { - final ProcessGroup group = rootGroup; - return group == null ? null : group.findControllerService(serviceIdentifier); + final ControllerServiceNode rootServiceNode = flowController.getRootControllerService(serviceIdentifier); + if (rootServiceNode != null) { + return rootServiceNode; + } + + return getRootGroup().findControllerService(serviceIdentifier); } + @Override - public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType, String groupId) { - ProcessGroup group = rootGroup; - if (group == null) { - return Collections.emptySet(); - } + public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType, final String groupId) { + final Set<ControllerServiceNode> serviceNodes; + if (groupId == null) { + serviceNodes = flowController.getRootControllerServices(); + } else { + ProcessGroup group = getRootGroup(); + if (!FlowController.ROOT_GROUP_ID_ALIAS.equals(groupId) && !group.getIdentifier().equals(groupId)) { + group = group.findProcessGroup(groupId); + } - if (!FlowController.ROOT_GROUP_ID_ALIAS.equals(groupId) && !group.getIdentifier().equals(groupId)) { - group = group.findProcessGroup(groupId); - } + if (group == null) { + return Collections.emptySet(); + } - if (group == null) { - return Collections.emptySet(); + serviceNodes = group.getControllerServices(true); } - final Set<ControllerServiceNode> serviceNodes = group.getControllerServices(true); - final Set<String> identifiers = new HashSet<>(); for (final ControllerServiceNode serviceNode : serviceNodes) { if (requireNonNull(serviceType).isAssignableFrom(serviceNode.getProxiedControllerService().getClass())) { @@ -579,7 +629,8 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi public void removeControllerService(final ControllerServiceNode serviceNode) { final ProcessGroup group = requireNonNull(serviceNode).getProcessGroup(); if (group == null) { - throw new IllegalArgumentException("Cannot remote Controller Service " + serviceNode + " because it does not belong to any Process Group"); + flowController.removeRootControllerService(serviceNode); + return; } group.removeControllerService(serviceNode); @@ -587,12 +638,11 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi @Override public Set<ControllerServiceNode> getAllControllerServices() { - final ProcessGroup group = rootGroup; - if (group == null) { - return Collections.emptySet(); - } + final Set<ControllerServiceNode> allServices = new HashSet<>(); + allServices.addAll(flowController.getRootControllerServices()); + allServices.addAll(getRootGroup().findAllControllerServices()); - return group.findAllControllerServices(); + return allServices; } /** @@ -709,4 +759,10 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi public void verifyCanStopReferencingComponents(final ControllerServiceNode serviceNode) { // we can always stop referencing components } + + + @Override + public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType) throws IllegalArgumentException { + throw new UnsupportedOperationException("Cannot obtain Controller Service Identifiers for service type " + serviceType + " without providing a Process Group Identifier"); + } } http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/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 a2a9104..8f20aab 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 @@ -186,7 +186,7 @@ public class StandardStateManagerProvider implements StateManagerProvider { provider.initialize(initContext); } - final ValidationContext validationContext = new StandardValidationContext(null, propertyStringMap, null, null); + final ValidationContext validationContext = new StandardValidationContext(null, propertyStringMap, null, null, null); final Collection<ValidationResult> results = provider.validate(validationContext); final StringBuilder validationFailures = new StringBuilder(); http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/ComponentSpecificControllerServiceLookup.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/ComponentSpecificControllerServiceLookup.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/ComponentSpecificControllerServiceLookup.java new file mode 100644 index 0000000..a08d118 --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/ComponentSpecificControllerServiceLookup.java @@ -0,0 +1,66 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.nifi.processor; + +import java.util.Set; + +import org.apache.nifi.controller.ControllerService; +import org.apache.nifi.controller.ControllerServiceLookup; +import org.apache.nifi.controller.service.ControllerServiceProvider; + +public class ComponentSpecificControllerServiceLookup implements ControllerServiceLookup { + private final ControllerServiceProvider serviceProvider; + private final String componentId; + private final String groupId; + + public ComponentSpecificControllerServiceLookup(final ControllerServiceProvider serviceProvider, final String componentId, final String groupId) { + this.serviceProvider = serviceProvider; + this.componentId = componentId; + this.groupId = groupId; + } + + @Override + public ControllerService getControllerService(final String serviceIdentifier) { + return serviceProvider.getControllerServiceForComponent(serviceIdentifier, componentId); + } + + @Override + public boolean isControllerServiceEnabled(final String serviceIdentifier) { + return serviceProvider.isControllerServiceEnabled(serviceIdentifier); + } + + @Override + public boolean isControllerServiceEnabling(final String serviceIdentifier) { + return serviceProvider.isControllerServiceEnabling(serviceIdentifier); + } + + @Override + public boolean isControllerServiceEnabled(final ControllerService service) { + return serviceProvider.isControllerServiceEnabled(service); + } + + @Override + public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType) throws IllegalArgumentException { + return serviceProvider.getControllerServiceIdentifiers(serviceType, groupId); + } + + @Override + public String getControllerServiceName(final String serviceIdentifier) { + return serviceProvider.getControllerServiceName(serviceIdentifier); + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/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 d83e08c..3c5acbb 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 @@ -101,7 +101,7 @@ public class StandardProcessContext implements ProcessContext, ControllerService @Override public ControllerService getControllerService(final String serviceIdentifier) { - return controllerServiceProvider.getControllerService(serviceIdentifier); + return controllerServiceProvider.getControllerServiceForComponent(serviceIdentifier, procNode.getIdentifier()); } @Override @@ -130,11 +130,11 @@ public class StandardProcessContext implements ProcessContext, ControllerService } @Override - public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType, String groupId) { + public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType) { if (!serviceType.isInterface()) { throw new IllegalArgumentException("ControllerServices may be referenced only via their interfaces; " + serviceType + " is not an interface"); } - return controllerServiceProvider.getControllerServiceIdentifiers(serviceType, groupId); + return controllerServiceProvider.getControllerServiceIdentifiers(serviceType, procNode.getProcessGroup().getIdentifier()); } @Override @@ -197,8 +197,8 @@ public class StandardProcessContext implements ProcessContext, ControllerService } @Override - public boolean hasConnection(Relationship relationship) { - Set<Connection> connections = procNode.getConnections(relationship); + public boolean hasConnection(final Relationship relationship) { + final Set<Connection> connections = procNode.getConnections(relationship); return connections != null && !connections.isEmpty(); } http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/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 1aec203..3fb8a21 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 @@ -45,22 +45,26 @@ public class StandardValidationContext implements ValidationContext { private final String annotationData; private final Set<String> serviceIdentifiersToNotValidate; private final String groupId; + private final String componentId; - public StandardValidationContext(final ControllerServiceProvider controllerServiceProvider, final Map<PropertyDescriptor, String> properties, final String annotationData, final String groupId) { - this(controllerServiceProvider, Collections.<String> emptySet(), properties, annotationData, groupId); + public StandardValidationContext(final ControllerServiceProvider controllerServiceProvider, final Map<PropertyDescriptor, String> properties, + final String annotationData, final String groupId, final String componentId) { + this(controllerServiceProvider, Collections.<String> emptySet(), properties, annotationData, groupId, componentId); } public StandardValidationContext( final ControllerServiceProvider controllerServiceProvider, final Set<String> serviceIdentifiersToNotValidate, final Map<PropertyDescriptor, String> properties, - final String annotationData, - final String groupId) { + final String annotationData, + final String groupId, + final String componentId) { this.controllerServiceProvider = controllerServiceProvider; this.properties = new HashMap<>(properties); this.annotationData = annotationData; this.serviceIdentifiersToNotValidate = serviceIdentifiersToNotValidate; this.groupId = groupId; + this.componentId = componentId; preparedQueries = new HashMap<>(properties.size()); for (final Map.Entry<PropertyDescriptor, String> entry : properties.entrySet()) { @@ -93,7 +97,8 @@ public class StandardValidationContext implements ValidationContext { @Override public ValidationContext getControllerServiceValidationContext(final ControllerService controllerService) { final ControllerServiceNode serviceNode = controllerServiceProvider.getControllerServiceNode(controllerService.getIdentifier()); - return new StandardValidationContext(controllerServiceProvider, serviceNode.getProperties(), serviceNode.getAnnotationData(), serviceNode.getProcessGroup().getIdentifier()); + return new StandardValidationContext(controllerServiceProvider, serviceNode.getProperties(), serviceNode.getAnnotationData(), + serviceNode.getProcessGroup().getIdentifier(), serviceNode.getIdentifier()); } @Override @@ -114,7 +119,7 @@ public class StandardValidationContext implements ValidationContext { @Override public ControllerServiceLookup getControllerServiceLookup() { - return controllerServiceProvider; + return new ComponentSpecificControllerServiceLookup(controllerServiceProvider, componentId, groupId); } @Override http://git-wip-us.apache.org/repos/asf/nifi/blob/c955ec16/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 9c8475e..1c52e17 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 @@ -33,13 +33,13 @@ public class StandardValidationContextFactory implements ValidationContextFactor } @Override - public ValidationContext newValidationContext(final Map<PropertyDescriptor, String> properties, final String annotationData, final String groupId) { - return new StandardValidationContext(serviceProvider, properties, annotationData, groupId); + public ValidationContext newValidationContext(final Map<PropertyDescriptor, String> properties, final String annotationData, final String groupId, final String componentId) { + return new StandardValidationContext(serviceProvider, properties, annotationData, groupId, componentId); } @Override public ValidationContext newValidationContext(final Set<String> serviceIdentifiersToNotValidate, - final Map<PropertyDescriptor, String> properties, final String annotationData, final String groupId) { - return new StandardValidationContext(serviceProvider, serviceIdentifiersToNotValidate, properties, annotationData, groupId); + final Map<PropertyDescriptor, String> properties, final String annotationData, final String groupId, String componentId) { + return new StandardValidationContext(serviceProvider, serviceIdentifiersToNotValidate, properties, annotationData, groupId, componentId); } }
