Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

NIFI-2033: Allow Controller Services to be scoped at Controller level instead of just group level. This closes #540

This commit is contained in:
Mark Payne 2016-06-17 09:08:44 -04:00 committed by Matt Gilman
parent 2c69c25323
commit c955ec1689
35 changed files with 826 additions and 554 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
nifi-api/src/main/java/org/apache/nifi/components/PropertyDescriptor.java
View File

@ -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);
}

3
nifi-api/src/main/java/org/apache/nifi/controller/ControllerServiceLookup.java
View File

@ -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

3
nifi-api/src/main/java/org/apache/nifi/web/NiFiWebConfigurationContext.java
View File

@ -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

6
nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceLookup.java
View File

@ -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;
}

12
nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java
View File

@ -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);
}
}

6
nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java
View File

@ -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);
}

14
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-documentation/src/main/java/org/apache/nifi/documentation/mock/MockControllerServiceLookup.java
View File

@ -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;
}
}

4
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/AbstractConfiguredComponent.java
View File

@ -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()) {

5
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ValidationContextFactory.java
View File

@ -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);
}

22
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java
View File

@ -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);
}

73
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
View File

@ -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 {
@ -2874,11 +2872,68 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
return controllerServiceProvider.getControllerService(serviceIdentifier);
}
@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);
}

4
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
View File

@ -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()));
}

28
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
View File

@ -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();

4
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingContext.java
View File

@ -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

4
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingInitializationContext.java
View File

@ -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

6
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/StandardFlowSerializer.java
View File

@ -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);
}

15
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/ControllerServiceLoader.java
View File

@ -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();

6
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java
View File

@ -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);
}

110
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
View File

@ -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");
}
}

2
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/manager/StandardStateManagerProvider.java
View File

@ -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();

66
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/ComponentSpecificControllerServiceLookup.java Normal file
View File

@ -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);
}
}

10
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessContext.java
View File

@ -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();
}

17
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContext.java
View File

@ -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

8
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContextFactory.java
View File

@ -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);
}
}

81
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java
View File

@ -38,11 +38,13 @@ import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.state.StateManagerProvider;
import org.apache.nifi.controller.AbstractControllerService;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.controller.FlowController;
import org.apache.nifi.controller.ProcessScheduler;
import org.apache.nifi.controller.ProcessorNode;
import org.apache.nifi.controller.ReportingTaskNode;
import org.apache.nifi.controller.StandardProcessorNode;
import org.apache.nifi.controller.ValidationContextFactory;
import org.apache.nifi.controller.cluster.Heartbeater;
import org.apache.nifi.controller.reporting.StandardReportingInitializationContext;
import org.apache.nifi.controller.reporting.StandardReportingTaskNode;
import org.apache.nifi.controller.service.ControllerServiceNode;
@ -73,7 +75,9 @@ public class TestStandardProcessScheduler {
private StandardProcessScheduler scheduler = null;
private ReportingTaskNode taskNode = null;
private TestReportingTask reportingTask = null;
private StateManagerProvider stateMgrProvider = Mockito.mock(StateManagerProvider.class);
private final StateManagerProvider stateMgrProvider = Mockito.mock(StateManagerProvider.class);
private FlowController controller;
private ProcessGroup rootGroup;
@Before
public void setup() throws InitializationException {
@ -89,6 +93,10 @@ public class TestStandardProcessScheduler {
final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(null);
taskNode = new StandardReportingTaskNode(reportingTask, UUID.randomUUID().toString(), null, scheduler, validationContextFactory);
controller = Mockito.mock(FlowController.class);
rootGroup = new MockProcessGroup();
Mockito.when(controller.getGroup(Mockito.anyString())).thenReturn(rootGroup);
}
/**
@ -115,20 +123,17 @@ public class TestStandardProcessScheduler {
assertTrue("After unscheduling Reporting Task, task ran an additional " + attemptsAfterStop + " times", attemptsAfterStop <= 1);
}
@Test(timeout = 6000)
@Test(timeout = 60000)
public void testDisableControllerServiceWithProcessorTryingToStartUsingIt() throws InterruptedException {
final Processor proc = new ServiceReferencingProcessor();
final ProcessGroup group = new MockProcessGroup();
final StandardControllerServiceProvider serviceProvider = new StandardControllerServiceProvider(scheduler, null, Mockito.mock(StateManagerProvider.class));
serviceProvider.setRootProcessGroup(group);
final StandardControllerServiceProvider serviceProvider = new StandardControllerServiceProvider(controller, scheduler, null, Mockito.mock(StateManagerProvider.class));
final ControllerServiceNode service = serviceProvider.createControllerService(NoStartServiceImpl.class.getName(), "service", true);
group.addControllerService(service);
rootGroup.addControllerService(service);
final ProcessorNode procNode = new StandardProcessorNode(proc, UUID.randomUUID().toString(),
new StandardValidationContextFactory(serviceProvider), scheduler, serviceProvider);
group.addProcessor(procNode);
rootGroup.addProcessor(procNode);
procNode.setProperty(ServiceReferencingProcessor.SERVICE_DESC.getName(), service.getIdentifier());
@ -184,16 +189,16 @@ public class TestStandardProcessScheduler {
}
@Override
public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
}
}
private void refreshNiFiProperties() {
try {
Field instanceField = NiFiProperties.class.getDeclaredField("instance");
final Field instanceField = NiFiProperties.class.getDeclaredField("instance");
instanceField.setAccessible(true);
instanceField.set(null, null);
} catch (Exception e) {
} catch (final Exception e) {
throw new IllegalStateException(e);
}
}
@ -206,12 +211,12 @@ public class TestStandardProcessScheduler {
@Test
public void validateServiceEnablementLogicHappensOnlyOnce() throws Exception {
final ProcessScheduler scheduler = createScheduler();
StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider);
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider);
final ControllerServiceNode serviceNode = provider.createControllerService(SimpleTestService.class.getName(),
"1", false);
assertFalse(serviceNode.isActive());
SimpleTestService ts = (SimpleTestService) serviceNode.getControllerServiceImplementation();
ExecutorService executor = Executors.newCachedThreadPool();
final SimpleTestService ts = (SimpleTestService) serviceNode.getControllerServiceImplementation();
final ExecutorService executor = Executors.newCachedThreadPool();
final AtomicBoolean asyncFailed = new AtomicBoolean();
for (int i = 0; i < 1000; i++) {
@ -221,7 +226,7 @@ public class TestStandardProcessScheduler {
try {
scheduler.enableControllerService(serviceNode);
assertTrue(serviceNode.isActive());
} catch (Exception e) {
} catch (final Exception e) {
e.printStackTrace();
asyncFailed.set(true);
}
@ -245,11 +250,11 @@ public class TestStandardProcessScheduler {
@Test
public void validateDisabledServiceCantBeDisabled() throws Exception {
final ProcessScheduler scheduler = createScheduler();
StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider);
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider);
final ControllerServiceNode serviceNode = provider.createControllerService(SimpleTestService.class.getName(),
"1", false);
SimpleTestService ts = (SimpleTestService) serviceNode.getControllerServiceImplementation();
ExecutorService executor = Executors.newCachedThreadPool();
final SimpleTestService ts = (SimpleTestService) serviceNode.getControllerServiceImplementation();
final ExecutorService executor = Executors.newCachedThreadPool();
final AtomicBoolean asyncFailed = new AtomicBoolean();
for (int i = 0; i < 1000; i++) {
@ -259,7 +264,7 @@ public class TestStandardProcessScheduler {
try {
scheduler.disableControllerService(serviceNode);
assertFalse(serviceNode.isActive());
} catch (Exception e) {
} catch (final Exception e) {
e.printStackTrace();
asyncFailed.set(true);
}
@ -283,13 +288,13 @@ public class TestStandardProcessScheduler {
@Test
public void validateEnabledServiceCanOnlyBeDisabledOnce() throws Exception {
final ProcessScheduler scheduler = createScheduler();
StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider);
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider);
final ControllerServiceNode serviceNode = provider.createControllerService(SimpleTestService.class.getName(),
"1", false);
SimpleTestService ts = (SimpleTestService) serviceNode.getControllerServiceImplementation();
final SimpleTestService ts = (SimpleTestService) serviceNode.getControllerServiceImplementation();
scheduler.enableControllerService(serviceNode);
assertTrue(serviceNode.isActive());
ExecutorService executor = Executors.newCachedThreadPool();
final ExecutorService executor = Executors.newCachedThreadPool();
final AtomicBoolean asyncFailed = new AtomicBoolean();
for (int i = 0; i < 1000; i++) {
@ -299,7 +304,7 @@ public class TestStandardProcessScheduler {
try {
scheduler.disableControllerService(serviceNode);
assertFalse(serviceNode.isActive());
} catch (Exception e) {
} catch (final Exception e) {
e.printStackTrace();
asyncFailed.set(true);
}
@ -317,7 +322,7 @@ public class TestStandardProcessScheduler {
@Test
public void validateDisablingOfTheFailedService() throws Exception {
final ProcessScheduler scheduler = createScheduler();
StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider);
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider);
final ControllerServiceNode serviceNode = provider.createControllerService(FailingService.class.getName(),
"1", false);
scheduler.enableControllerService(serviceNode);
@ -348,8 +353,8 @@ public class TestStandardProcessScheduler {
@Test
public void validateEnabledDisableMultiThread() throws Exception {
final ProcessScheduler scheduler = createScheduler();
StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider);
ExecutorService executor = Executors.newCachedThreadPool();
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider);
final ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 200; i++) {
final ControllerServiceNode serviceNode = provider
.createControllerService(RandomShortDelayEnablingService.class.getName(), "1", false);
@ -391,10 +396,10 @@ public class TestStandardProcessScheduler {
@Test
public void validateNeverEnablingServiceCanStillBeDisabled() throws Exception {
final ProcessScheduler scheduler = createScheduler();
StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider);
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider);
final ControllerServiceNode serviceNode = provider.createControllerService(LongEnablingService.class.getName(),
"1", false);
LongEnablingService ts = (LongEnablingService) serviceNode.getControllerServiceImplementation();
final LongEnablingService ts = (LongEnablingService) serviceNode.getControllerServiceImplementation();
ts.setLimit(Long.MAX_VALUE);
scheduler.enableControllerService(serviceNode);
Thread.sleep(100);
@ -416,10 +421,10 @@ public class TestStandardProcessScheduler {
@Test
public void validateLongEnablingServiceCanStillBeDisabled() throws Exception {
final ProcessScheduler scheduler = createScheduler();
StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider);
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateMgrProvider);
final ControllerServiceNode serviceNode = provider.createControllerService(LongEnablingService.class.getName(),
"1", false);
LongEnablingService ts = (LongEnablingService) serviceNode.getControllerServiceImplementation();
final LongEnablingService ts = (LongEnablingService) serviceNode.getControllerServiceImplementation();
ts.setLimit(3000);
scheduler.enableControllerService(serviceNode);
Thread.sleep(2000);
@ -440,7 +445,7 @@ public class TestStandardProcessScheduler {
public static class FailingService extends AbstractControllerService {
@OnEnabled
public void enable(ConfigurationContext context) {
public void enable(final ConfigurationContext context) {
throw new RuntimeException("intentional");
}
}
@ -449,10 +454,10 @@ public class TestStandardProcessScheduler {
private final Random random = new Random();
@OnEnabled
public void enable(ConfigurationContext context) {
public void enable(final ConfigurationContext context) {
try {
Thread.sleep(random.nextInt(20));
} catch (InterruptedException e) {
} catch (final InterruptedException e) {
Thread.currentThread().interrupt();
}
}
@ -464,12 +469,12 @@ public class TestStandardProcessScheduler {
private final AtomicInteger disableCounter = new AtomicInteger();
@OnEnabled
public void enable(ConfigurationContext context) {
public void enable(final ConfigurationContext context) {
this.enableCounter.incrementAndGet();
}
@OnDisabled
public void disable(ConfigurationContext context) {
public void disable(final ConfigurationContext context) {
this.disableCounter.incrementAndGet();
}
@ -489,13 +494,13 @@ public class TestStandardProcessScheduler {
private volatile long limit;
@OnEnabled
public void enable(ConfigurationContext context) throws Exception {
public void enable(final ConfigurationContext context) throws Exception {
this.enableCounter.incrementAndGet();
Thread.sleep(limit);
}
@OnDisabled
public void disable(ConfigurationContext context) {
public void disable(final ConfigurationContext context) {
this.disableCounter.incrementAndGet();
}
@ -507,7 +512,7 @@ public class TestStandardProcessScheduler {
return this.disableCounter.get();
}
public void setLimit(long limit) {
public void setLimit(final long limit) {
this.limit = limit;
}
}

2
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java
View File

@ -46,7 +46,7 @@ public class StandardControllerServiceProviderTest {
public void setup() throws Exception {
String id = "id";
String clazz = "org.apache.nifi.controller.service.util.TestControllerService";
ControllerServiceProvider provider = new StandardControllerServiceProvider(null, null, new StateManagerProvider() {
ControllerServiceProvider provider = new StandardControllerServiceProvider(null, null, null, new StateManagerProvider() {
@Override
public StateManager getStateManager(final String componentId) {
return Mockito.mock(StateManager.class);

41
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java
View File

@ -29,6 +29,7 @@ import java.util.UUID;
import org.apache.nifi.components.state.StateManager;
import org.apache.nifi.components.state.StateManagerProvider;
import org.apache.nifi.controller.FlowController;
import org.apache.nifi.controller.ProcessScheduler;
import org.apache.nifi.controller.ProcessorNode;
import org.apache.nifi.controller.ScheduledState;
@ -49,7 +50,7 @@ import org.mockito.Mockito;
public class TestStandardControllerServiceProvider {
private static StateManagerProvider stateManagerProvider = new StateManagerProvider() {
@Override
public StateManager getStateManager(String componentId) {
public StateManager getStateManager(final String componentId) {
return Mockito.mock(StateManager.class);
}
@ -66,7 +67,7 @@ public class TestStandardControllerServiceProvider {
}
@Override
public void onComponentRemoved(String componentId) {
public void onComponentRemoved(final String componentId) {
}
};
@ -81,20 +82,26 @@ public class TestStandardControllerServiceProvider {
@Test
public void testDisableControllerService() {
final ProcessGroup procGroup = new MockProcessGroup();
final FlowController controller = Mockito.mock(FlowController.class);
Mockito.when(controller.getGroup(Mockito.anyString())).thenReturn(procGroup);
final ProcessScheduler scheduler = createScheduler();
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider);
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateManagerProvider);
final ControllerServiceNode serviceNode = provider.createControllerService(ServiceB.class.getName(), "B", false);
provider.enableControllerService(serviceNode);
provider.disableControllerService(serviceNode);
}
@Test(timeout = 1000000)
@Test(timeout = 10000)
public void testEnableDisableWithReference() {
final ProcessScheduler scheduler = createScheduler();
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider);
final ProcessGroup group = new MockProcessGroup();
provider.setRootProcessGroup(group);
final FlowController controller = Mockito.mock(FlowController.class);
Mockito.when(controller.getGroup(Mockito.anyString())).thenReturn(group);
final ProcessScheduler scheduler = createScheduler();
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateManagerProvider);
final ControllerServiceNode serviceNodeB = provider.createControllerService(ServiceB.class.getName(), "B", false);
final ControllerServiceNode serviceNodeA = provider.createControllerService(ServiceA.class.getName(), "A", false);
@ -148,10 +155,12 @@ public class TestStandardControllerServiceProvider {
}
}
public void testEnableReferencingServicesGraph(ProcessScheduler scheduler) {
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider);
public void testEnableReferencingServicesGraph(final ProcessScheduler scheduler) {
final ProcessGroup procGroup = new MockProcessGroup();
provider.setRootProcessGroup(procGroup);
final FlowController controller = Mockito.mock(FlowController.class);
Mockito.when(controller.getGroup(Mockito.anyString())).thenReturn(procGroup);
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null, stateManagerProvider);
// build a graph of controller services with dependencies as such:
//
@ -199,7 +208,11 @@ public class TestStandardControllerServiceProvider {
@Test
public void testOrderingOfServices() {
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(null, null, stateManagerProvider);
final ProcessGroup procGroup = new MockProcessGroup();
final FlowController controller = Mockito.mock(FlowController.class);
Mockito.when(controller.getGroup(Mockito.anyString())).thenReturn(procGroup);
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, null, null, stateManagerProvider);
final ControllerServiceNode serviceNode1 = provider.createControllerService(ServiceA.class.getName(), "1", false);
final ControllerServiceNode serviceNode2 = provider.createControllerService(ServiceB.class.getName(), "2", false);
@ -354,8 +367,12 @@ public class TestStandardControllerServiceProvider {
@Test
public void testEnableReferencingComponents() {
final ProcessGroup procGroup = new MockProcessGroup();
final FlowController controller = Mockito.mock(FlowController.class);
Mockito.when(controller.getGroup(Mockito.anyString())).thenReturn(procGroup);
final StandardProcessScheduler scheduler = createScheduler();
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(null, null, stateManagerProvider);
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, null, null, stateManagerProvider);
final ControllerServiceNode serviceNode = provider.createControllerService(ServiceA.class.getName(), "1", false);
final ProcessorNode procNode = createProcessor(scheduler, provider);

169
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/mock/MockProcessGroup.java
View File

@ -17,6 +17,13 @@
package org.apache.nifi.controller.service.mock;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.nifi.authorization.Resource;
import org.apache.nifi.authorization.resource.Authorizable;
import org.apache.nifi.connectable.Connectable;
@ -34,14 +41,9 @@ import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.groups.ProcessGroupCounts;
import org.apache.nifi.groups.RemoteProcessGroup;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
public class MockProcessGroup implements ProcessGroup {
private Map<String, ControllerServiceNode> serviceMap = new HashMap<>();
private final Map<String, ControllerServiceNode> serviceMap = new HashMap<>();
private final Map<String, ProcessorNode> processorMap = new HashMap<>();
@Override
public Authorizable getParentAuthorizable() {
@ -59,7 +61,7 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public void setParent(ProcessGroup group) {
public void setParent(final ProcessGroup group) {
}
@ -74,12 +76,12 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public void setName(String name) {
public void setName(final String name) {
}
@Override
public void setPosition(Position position) {
public void setPosition(final Position position) {
}
@ -94,7 +96,7 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public void setComments(String comments) {
public void setComments(final String comments) {
}
@ -114,67 +116,67 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public void enableProcessor(ProcessorNode processor) {
public void enableProcessor(final ProcessorNode processor) {
}
@Override
public void enableInputPort(Port port) {
public void enableInputPort(final Port port) {
}
@Override
public void enableOutputPort(Port port) {
public void enableOutputPort(final Port port) {
}
@Override
public void startProcessor(ProcessorNode processor) {
public void startProcessor(final ProcessorNode processor) {
}
@Override
public void startInputPort(Port port) {
public void startInputPort(final Port port) {
}
@Override
public void startOutputPort(Port port) {
public void startOutputPort(final Port port) {
}
@Override
public void startFunnel(Funnel funnel) {
public void startFunnel(final Funnel funnel) {
}
@Override
public void stopProcessor(ProcessorNode processor) {
public void stopProcessor(final ProcessorNode processor) {
}
@Override
public void stopInputPort(Port port) {
public void stopInputPort(final Port port) {
}
@Override
public void stopOutputPort(Port port) {
public void stopOutputPort(final Port port) {
}
@Override
public void disableProcessor(ProcessorNode processor) {
public void disableProcessor(final ProcessorNode processor) {
}
@Override
public void disableInputPort(Port port) {
public void disableInputPort(final Port port) {
}
@Override
public void disableOutputPort(Port port) {
public void disableOutputPort(final Port port) {
}
@ -189,12 +191,12 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public void addInputPort(Port port) {
public void addInputPort(final Port port) {
}
@Override
public void removeInputPort(Port port) {
public void removeInputPort(final Port port) {
}
@ -204,22 +206,22 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public Port getInputPort(String id) {
public Port getInputPort(final String id) {
return null;
}
@Override
public void addOutputPort(Port port) {
public void addOutputPort(final Port port) {
}
@Override
public void removeOutputPort(Port port) {
public void removeOutputPort(final Port port) {
}
@Override
public Port getOutputPort(String id) {
public Port getOutputPort(final String id) {
return null;
}
@ -229,12 +231,12 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public void addProcessGroup(ProcessGroup group) {
public void addProcessGroup(final ProcessGroup group) {
}
@Override
public ProcessGroup getProcessGroup(String id) {
public ProcessGroup getProcessGroup(final String id) {
return null;
}
@ -244,28 +246,29 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public void removeProcessGroup(ProcessGroup group) {
public void removeProcessGroup(final ProcessGroup group) {
}
@Override
public void addProcessor(ProcessorNode processor) {
public void addProcessor(final ProcessorNode processor) {
processor.setProcessGroup(this);
processorMap.put(processor.getIdentifier(), processor);
}
@Override
public void removeProcessor(ProcessorNode processor) {
public void removeProcessor(final ProcessorNode processor) {
processorMap.remove(processor.getIdentifier());
}
@Override
public Set<ProcessorNode> getProcessors() {
return null;
return new HashSet<>(processorMap.values());
}
@Override
public ProcessorNode getProcessor(String id) {
return null;
public ProcessorNode getProcessor(final String id) {
return processorMap.get(id);
}
@Override
@ -274,27 +277,27 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public Connectable getConnectable(String id) {
public Connectable getConnectable(final String id) {
return null;
}
@Override
public void addConnection(Connection connection) {
public void addConnection(final Connection connection) {
}
@Override
public void removeConnection(Connection connection) {
public void removeConnection(final Connection connection) {
}
@Override
public void inheritConnection(Connection connection) {
public void inheritConnection(final Connection connection) {
}
@Override
public Connection getConnection(String id) {
public Connection getConnection(final String id) {
return null;
}
@ -304,7 +307,7 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public Connection findConnection(String id) {
public Connection findConnection(final String id) {
return null;
}
@ -314,12 +317,12 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public Funnel findFunnel(String id) {
public Funnel findFunnel(final String id) {
return null;
}
@Override
public ControllerServiceNode findControllerService(String id) {
public ControllerServiceNode findControllerService(final String id) {
return serviceMap.get(id);
}
@ -329,17 +332,17 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public void addRemoteProcessGroup(RemoteProcessGroup remoteGroup) {
public void addRemoteProcessGroup(final RemoteProcessGroup remoteGroup) {
}
@Override
public void removeRemoteProcessGroup(RemoteProcessGroup remoteGroup) {
public void removeRemoteProcessGroup(final RemoteProcessGroup remoteGroup) {
}
@Override
public RemoteProcessGroup getRemoteProcessGroup(String id) {
public RemoteProcessGroup getRemoteProcessGroup(final String id) {
return null;
}
@ -349,12 +352,12 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public void addLabel(Label label) {
public void addLabel(final Label label) {
}
@Override
public void removeLabel(Label label) {
public void removeLabel(final Label label) {
}
@ -364,12 +367,12 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public Label getLabel(String id) {
public Label getLabel(final String id) {
return null;
}
@Override
public ProcessGroup findProcessGroup(String id) {
public ProcessGroup findProcessGroup(final String id) {
return null;
}
@ -379,7 +382,7 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public RemoteProcessGroup findRemoteProcessGroup(String id) {
public RemoteProcessGroup findRemoteProcessGroup(final String id) {
return null;
}
@ -389,17 +392,17 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public ProcessorNode findProcessor(String id) {
return null;
public ProcessorNode findProcessor(final String id) {
return processorMap.get(id);
}
@Override
public List<ProcessorNode> findAllProcessors() {
return null;
return new ArrayList<>(processorMap.values());
}
@Override
public Label findLabel(String id) {
public Label findLabel(final String id) {
return null;
}
@ -409,7 +412,7 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public Port findInputPort(String id) {
public Port findInputPort(final String id) {
return null;
}
@ -419,12 +422,12 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public Port getInputPortByName(String name) {
public Port getInputPortByName(final String name) {
return null;
}
@Override
public Port findOutputPort(String id) {
public Port findOutputPort(final String id) {
return null;
}
@ -434,17 +437,17 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public Port getOutputPortByName(String name) {
public Port getOutputPortByName(final String name) {
return null;
}
@Override
public void addFunnel(Funnel funnel) {
public void addFunnel(final Funnel funnel) {
}
@Override
public void addFunnel(Funnel funnel, boolean autoStart) {
public void addFunnel(final Funnel funnel, final boolean autoStart) {
}
@ -454,33 +457,33 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public Funnel getFunnel(String id) {
public Funnel getFunnel(final String id) {
return null;
}
@Override
public void removeFunnel(Funnel funnel) {
public void removeFunnel(final Funnel funnel) {
}
@Override
public void addControllerService(ControllerServiceNode service) {
public void addControllerService(final ControllerServiceNode service) {
serviceMap.put(service.getIdentifier(), service);
service.setProcessGroup(this);
}
@Override
public ControllerServiceNode getControllerService(String id) {
public ControllerServiceNode getControllerService(final String id) {
return serviceMap.get(id);
}
@Override
public Set<ControllerServiceNode> getControllerServices(boolean recursive) {
public Set<ControllerServiceNode> getControllerServices(final boolean recursive) {
return new HashSet<>(serviceMap.values());
}
@Override
public void removeControllerService(ControllerServiceNode service) {
public void removeControllerService(final ControllerServiceNode service) {
serviceMap.remove(service.getIdentifier());
}
@ -490,17 +493,17 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public void remove(Snippet snippet) {
public void remove(final Snippet snippet) {
}
@Override
public Connectable findConnectable(String identifier) {
public Connectable findConnectable(final String identifier) {
return null;
}
@Override
public void move(Snippet snippet, ProcessGroup destination) {
public void move(final Snippet snippet, final ProcessGroup destination) {
}
@ -510,7 +513,7 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public void verifyCanDelete(boolean ignorePortConnections) {
public void verifyCanDelete(final boolean ignorePortConnections) {
}
@ -525,31 +528,31 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public void verifyCanDelete(Snippet snippet) {
public void verifyCanDelete(final Snippet snippet) {
}
@Override
public void verifyCanMove(Snippet snippet, ProcessGroup newProcessGroup) {
public void verifyCanMove(final Snippet snippet, final ProcessGroup newProcessGroup) {
}
@Override
public void addTemplate(Template template) {
public void addTemplate(final Template template) {
throw new UnsupportedOperationException();
}
@Override
public void removeTemplate(Template template) {
public void removeTemplate(final Template template) {
}
@Override
public Template getTemplate(String id) {
public Template getTemplate(final String id) {
return null;
}
@Override
public Template findTemplate(String id) {
public Template findTemplate(final String id) {
return null;
}
@ -564,10 +567,10 @@ public class MockProcessGroup implements ProcessGroup {
}
@Override
public void verifyCanStart(Connectable connectable) {
public void verifyCanStart(final Connectable connectable) {
}
@Override
public void verifyCanStop(Connectable connectable) {
public void verifyCanStop(final Connectable connectable) {
}
}

10
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/processor/TestStandardPropertyValue.java
View File

@ -148,27 +148,27 @@ public class TestStandardPropertyValue {
}
@Override
public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType, String groupId) {
public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType) {
return null;
}
@Override
public boolean isControllerServiceEnabled(String serviceIdentifier) {
public boolean isControllerServiceEnabled(final String serviceIdentifier) {
return true;
}
@Override
public boolean isControllerServiceEnabled(ControllerService service) {
public boolean isControllerServiceEnabled(final ControllerService service) {
return true;
}
@Override
public String getControllerServiceName(String serviceIdentifier) {
public String getControllerServiceName(final String serviceIdentifier) {
return null;
}
@Override
public boolean isControllerServiceEnabling(String serviceIdentifier) {
public boolean isControllerServiceEnabling(final String serviceIdentifier) {
return false;
}
}

404
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
View File

File diff suppressed because it is too large Load Diff

114
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiWebConfigurationContext.java
View File

@ -16,22 +16,7 @@
*/
package org.apache.nifi.web;
import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URLEncoder;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
import javax.ws.rs.HttpMethod;
import javax.ws.rs.core.MultivaluedMap;
import javax.ws.rs.core.Response;
import com.sun.jersey.core.util.MultivaluedMapImpl;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.action.Action;
import org.apache.nifi.action.Component;
@ -56,8 +41,8 @@ import org.apache.nifi.cluster.coordination.http.replication.RequestReplicator;
import org.apache.nifi.cluster.manager.NodeResponse;
import org.apache.nifi.cluster.manager.exception.IllegalClusterStateException;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.ControllerServiceLookup;
import org.apache.nifi.controller.reporting.ReportingTaskProvider;
import org.apache.nifi.controller.service.ControllerServiceProvider;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.web.api.dto.ControllerServiceDTO;
import org.apache.nifi.web.api.dto.ProcessorConfigDTO;
@ -74,7 +59,20 @@ import org.slf4j.LoggerFactory;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder;
import com.sun.jersey.core.util.MultivaluedMapImpl;
import javax.ws.rs.HttpMethod;
import javax.ws.rs.core.MultivaluedMap;
import javax.ws.rs.core.Response;
import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URLEncoder;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
/**
* Implements the NiFiWebConfigurationContext interface to support a context in both standalone and clustered environments.
@ -88,7 +86,7 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
private NiFiServiceFacade serviceFacade;
private ClusterCoordinator clusterCoordinator;
private RequestReplicator requestReplicator;
private ControllerServiceLookup controllerServiceLookup;
private ControllerServiceProvider controllerServiceProvider;
private ReportingTaskProvider reportingTaskProvider;
private AuditService auditService;
private Authorizer authorizer;
@ -113,10 +111,10 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
}
@Override
public ControllerService getControllerService(String serviceIdentifier) {
public ControllerService getControllerService(final String serviceIdentifier, final String componentId) {
final NiFiUser user = NiFiUserUtils.getNiFiUser();
authorizeFlowAccess(user);
return controllerServiceLookup.getControllerService(serviceIdentifier);
return controllerServiceProvider.getControllerServiceForComponent(serviceIdentifier, componentId);
}
@Override
@ -193,7 +191,7 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
try {
// record the operations
auditService.addActions(actions);
} catch (Throwable t) {
} catch (final Throwable t) {
logger.warn("Unable to record actions: " + t.getMessage());
if (logger.isDebugEnabled()) {
logger.warn(StringUtils.EMPTY, t);
@ -329,21 +327,21 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
// create the request URL
URI requestUrl;
try {
String path = "/nifi-api/processors/" + URLEncoder.encode(id, "UTF-8");
final String path = "/nifi-api/processors/" + URLEncoder.encode(id, "UTF-8");
requestUrl = new URI(requestContext.getScheme(), null, "localhost", 0, path, null, null);
} catch (final URISyntaxException | UnsupportedEncodingException use) {
throw new ClusterRequestException(use);
}
// set the request parameters
MultivaluedMap<String, String> parameters = new MultivaluedMapImpl();
final MultivaluedMap<String, String> parameters = new MultivaluedMapImpl();
parameters.add(VERBOSE_PARAM, "true");
// replicate request
NodeResponse nodeResponse;
try {
nodeResponse = requestReplicator.replicate(HttpMethod.GET, requestUrl, parameters, getHeaders(requestContext)).awaitMergedResponse();
} catch (InterruptedException e) {
} catch (final InterruptedException e) {
throw new IllegalClusterStateException("Request was interrupted while waiting for response from node");
}
@ -381,28 +379,28 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
// create the request URL
URI requestUrl;
try {
String path = "/nifi-api/processors/" + URLEncoder.encode(id, "UTF-8");
final String path = "/nifi-api/processors/" + URLEncoder.encode(id, "UTF-8");
requestUrl = new URI(requestContext.getScheme(), null, "localhost", 0, path, null, null);
} catch (final URISyntaxException | UnsupportedEncodingException use) {
throw new ClusterRequestException(use);
}
// create the revision
RevisionDTO revisionDto = new RevisionDTO();
final RevisionDTO revisionDto = new RevisionDTO();
revisionDto.setClientId(revision.getClientId());
revisionDto.setVersion(revision.getVersion());
// create the processor entity
ProcessorEntity processorEntity = new ProcessorEntity();
final ProcessorEntity processorEntity = new ProcessorEntity();
processorEntity.setRevision(revisionDto);
// create the processor dto
ProcessorDTO processorDto = new ProcessorDTO();
final ProcessorDTO processorDto = new ProcessorDTO();
processorEntity.setComponent(processorDto);
processorDto.setId(id);
// create the processor configuration with the given annotation data
ProcessorConfigDTO configDto = new ProcessorConfigDTO();
final ProcessorConfigDTO configDto = new ProcessorConfigDTO();
processorDto.setConfig(configDto);
configDto.setAnnotationData(annotationData);
@ -414,7 +412,7 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
NodeResponse nodeResponse;
try {
nodeResponse = requestReplicator.replicate(HttpMethod.PUT, requestUrl, processorEntity, headers).awaitMergedResponse();
} catch (InterruptedException e) {
} catch (final InterruptedException e) {
throw new IllegalClusterStateException("Request was interrupted while waiting for response from node");
}
@ -474,7 +472,7 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
// if the lookup has the service that means we are either a node or
// the ncm and the service is available there only
if (controllerServiceLookup.getControllerService(id) != null) {
if (controllerServiceProvider.getControllerService(id) != null) {
controllerService = serviceFacade.getControllerService(id).getComponent();
} else {
// if this is a standalone instance the service should have been found above... there should
@ -486,20 +484,20 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
// create the request URL
URI requestUrl;
try {
String path = "/nifi-api/controller-services/node/" + URLEncoder.encode(id, "UTF-8");
final String path = "/nifi-api/controller-services/node/" + URLEncoder.encode(id, "UTF-8");
requestUrl = new URI(requestContext.getScheme(), null, "localhost", 0, path, null, null);
} catch (final URISyntaxException | UnsupportedEncodingException use) {
throw new ClusterRequestException(use);
}
// set the request parameters
MultivaluedMap<String, String> parameters = new MultivaluedMapImpl();
final MultivaluedMap<String, String> parameters = new MultivaluedMapImpl();
// replicate request
NodeResponse nodeResponse;
try {
nodeResponse = requestReplicator.replicate(HttpMethod.GET, requestUrl, parameters, getHeaders(requestContext)).awaitMergedResponse();
} catch (InterruptedException e) {
} catch (final InterruptedException e) {
throw new IllegalClusterStateException("Request was interrupted while waiting for response from node");
}
@ -531,7 +529,7 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
});
final ControllerServiceDTO controllerService;
if (controllerServiceLookup.getControllerService(id) != null) {
if (controllerServiceProvider.getControllerService(id) != null) {
final ControllerServiceDTO controllerServiceDto = new ControllerServiceDTO();
controllerServiceDto.setId(id);
controllerServiceDto.setAnnotationData(annotationData);
@ -562,23 +560,23 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
// create the request URL
URI requestUrl;
try {
String path = "/nifi-api/controller-services/node/" + URLEncoder.encode(id, "UTF-8");
final String path = "/nifi-api/controller-services/node/" + URLEncoder.encode(id, "UTF-8");
requestUrl = new URI(requestContext.getScheme(), null, "localhost", 0, path, null, null);
} catch (final URISyntaxException | UnsupportedEncodingException use) {
throw new ClusterRequestException(use);
}
// create the revision
RevisionDTO revisionDto = new RevisionDTO();
final RevisionDTO revisionDto = new RevisionDTO();
revisionDto.setClientId(revision.getClientId());
revisionDto.setVersion(revision.getVersion());
// create the controller service entity
ControllerServiceEntity controllerServiceEntity = new ControllerServiceEntity();
final ControllerServiceEntity controllerServiceEntity = new ControllerServiceEntity();
controllerServiceEntity.setRevision(revisionDto);
// create the controller service dto
ControllerServiceDTO controllerServiceDto = new ControllerServiceDTO();
final ControllerServiceDTO controllerServiceDto = new ControllerServiceDTO();
controllerServiceEntity.setComponent(controllerServiceDto);
controllerServiceDto.setId(id);
controllerServiceDto.setAnnotationData(annotationData);
@ -591,7 +589,7 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
NodeResponse nodeResponse;
try {
nodeResponse = requestReplicator.replicate(HttpMethod.PUT, requestUrl, controllerServiceEntity, headers).awaitMergedResponse();
} catch (InterruptedException e) {
} catch (final InterruptedException e) {
throw new IllegalClusterStateException("Request was interrupted while waiting for response from node");
}
@ -652,20 +650,20 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
// create the request URL
URI requestUrl;
try {
String path = "/nifi-api/reporting-tasks/node/" + URLEncoder.encode(id, "UTF-8");
final String path = "/nifi-api/reporting-tasks/node/" + URLEncoder.encode(id, "UTF-8");
requestUrl = new URI(requestContext.getScheme(), null, "localhost", 0, path, null, null);
} catch (final URISyntaxException | UnsupportedEncodingException use) {
throw new ClusterRequestException(use);
}
// set the request parameters
MultivaluedMap<String, String> parameters = new MultivaluedMapImpl();
final MultivaluedMap<String, String> parameters = new MultivaluedMapImpl();
// replicate request
NodeResponse nodeResponse;
try {
nodeResponse = requestReplicator.replicate(HttpMethod.GET, requestUrl, parameters, getHeaders(requestContext)).awaitMergedResponse();
} catch (InterruptedException e) {
} catch (final InterruptedException e) {
throw new IllegalClusterStateException("Request was interrupted while waiting for response from node");
}
@ -727,23 +725,23 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
// create the request URL
URI requestUrl;
try {
String path = "/nifi-api/reporting-tasks/node/" + URLEncoder.encode(id, "UTF-8");
final String path = "/nifi-api/reporting-tasks/node/" + URLEncoder.encode(id, "UTF-8");
requestUrl = new URI(requestContext.getScheme(), null, "localhost", 0, path, null, null);
} catch (final URISyntaxException | UnsupportedEncodingException use) {
throw new ClusterRequestException(use);
}
// create the revision
RevisionDTO revisionDto = new RevisionDTO();
final RevisionDTO revisionDto = new RevisionDTO();
revisionDto.setClientId(revision.getClientId());
revisionDto.setVersion(revision.getVersion());
// create the reporting task entity
ReportingTaskEntity reportingTaskEntity = new ReportingTaskEntity();
final ReportingTaskEntity reportingTaskEntity = new ReportingTaskEntity();
reportingTaskEntity.setRevision(revisionDto);
// create the reporting task dto
ReportingTaskDTO reportingTaskDto = new ReportingTaskDTO();
final ReportingTaskDTO reportingTaskDto = new ReportingTaskDTO();
reportingTaskEntity.setComponent(reportingTaskDto);
reportingTaskDto.setId(id);
reportingTaskDto.setAnnotationData(annotationData);
@ -756,7 +754,7 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
NodeResponse nodeResponse;
try {
nodeResponse = requestReplicator.replicate(HttpMethod.PUT, requestUrl, reportingTaskEntity, headers).awaitMergedResponse();
} catch (InterruptedException e) {
} catch (final InterruptedException e) {
throw new IllegalClusterStateException("Request was interrupted while waiting for response from node");
}
@ -831,35 +829,35 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
}
}
public void setClusterCoordinator(ClusterCoordinator clusterCoordinator) {
public void setClusterCoordinator(final ClusterCoordinator clusterCoordinator) {
this.clusterCoordinator = clusterCoordinator;
}
public void setRequestReplicator(RequestReplicator requestReplicator) {
public void setRequestReplicator(final RequestReplicator requestReplicator) {
this.requestReplicator = requestReplicator;
}
public void setProperties(NiFiProperties properties) {
public void setProperties(final NiFiProperties properties) {
this.properties = properties;
}
public void setServiceFacade(NiFiServiceFacade serviceFacade) {
public void setServiceFacade(final NiFiServiceFacade serviceFacade) {
this.serviceFacade = serviceFacade;
}
public void setAuditService(AuditService auditService) {
public void setAuditService(final AuditService auditService) {
this.auditService = auditService;
}
public void setControllerServiceLookup(ControllerServiceLookup controllerServiceLookup) {
this.controllerServiceLookup = controllerServiceLookup;
public void setControllerServiceProvider(final ControllerServiceProvider controllerServiceProvider) {
this.controllerServiceProvider = controllerServiceProvider;
}
public void setReportingTaskProvider(ReportingTaskProvider reportingTaskProvider) {
public void setReportingTaskProvider(final ReportingTaskProvider reportingTaskProvider) {
this.reportingTaskProvider = reportingTaskProvider;
}
public void setAuthorizer(Authorizer authorizer) {
public void setAuthorizer(final Authorizer authorizer) {
this.authorizer = authorizer;
}
}

51
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
View File

@ -169,7 +169,7 @@ public final class DtoFactory {
@SuppressWarnings("rawtypes")
private final static Comparator<Class> CLASS_NAME_COMPARATOR = new Comparator<Class>() {
@Override
public int compare(Class class1, Class class2) {
public int compare(final Class class1, final Class class2) {
return Collator.getInstance(Locale.US).compare(class1.getSimpleName(), class2.getSimpleName());
}
};
@ -297,8 +297,8 @@ public final class DtoFactory {
historyDto.setLastRefreshed(history.getLastRefreshed());
if (history.getActions() != null) {
List<ActionDTO> actionDtos = new ArrayList<>();
for (Action action : history.getActions()) {
final List<ActionDTO> actionDtos = new ArrayList<>();
for (final Action action : history.getActions()) {
actionDtos.add(createActionDto(action));
}
historyDto.setActions(actionDtos);
@ -998,7 +998,7 @@ public final class DtoFactory {
* @param originalSnippet snippet
* @return dto
*/
public FlowSnippetDTO copySnippetContents(FlowSnippetDTO originalSnippet) {
public FlowSnippetDTO copySnippetContents(final FlowSnippetDTO originalSnippet) {
final FlowSnippetDTO copySnippet = new FlowSnippetDTO();
if (originalSnippet.getConnections() != null) {
@ -1113,7 +1113,7 @@ public final class DtoFactory {
// sort a copy of the properties
final Map<PropertyDescriptor, String> sortedProperties = new TreeMap<>(new Comparator<PropertyDescriptor>() {
@Override
public int compare(PropertyDescriptor o1, PropertyDescriptor o2) {
public int compare(final PropertyDescriptor o1, final PropertyDescriptor o2) {
return Collator.getInstance(Locale.US).compare(o1.getName(), o2.getName());
}
});
@ -1124,7 +1124,7 @@ public final class DtoFactory {
final Map<PropertyDescriptor, String> orderedProperties = new LinkedHashMap<>();
final List<PropertyDescriptor> descriptors = reportingTask.getPropertyDescriptors();
if (descriptors != null && !descriptors.isEmpty()) {
for (PropertyDescriptor descriptor : descriptors) {
for (final PropertyDescriptor descriptor : descriptors) {
orderedProperties.put(descriptor, null);
}
}
@ -1137,7 +1137,7 @@ public final class DtoFactory {
final PropertyDescriptor descriptor = entry.getKey();
// store the property descriptor
dto.getDescriptors().put(descriptor.getName(), createPropertyDescriptorDto(descriptor, "root"));
dto.getDescriptors().put(descriptor.getName(), createPropertyDescriptorDto(descriptor, null));
// determine the property value - don't include sensitive properties
String propertyValue = entry.getValue();
@ -1177,7 +1177,7 @@ public final class DtoFactory {
// sort a copy of the properties
final Map<PropertyDescriptor, String> sortedProperties = new TreeMap<>(new Comparator<PropertyDescriptor>() {
@Override
public int compare(PropertyDescriptor o1, PropertyDescriptor o2) {
public int compare(final PropertyDescriptor o1, final PropertyDescriptor o2) {
return Collator.getInstance(Locale.US).compare(o1.getName(), o2.getName());
}
});
@ -1188,7 +1188,7 @@ public final class DtoFactory {
final Map<PropertyDescriptor, String> orderedProperties = new LinkedHashMap<>();
final List<PropertyDescriptor> descriptors = controllerService.getPropertyDescriptors();
if (descriptors != null && !descriptors.isEmpty()) {
for (PropertyDescriptor descriptor : descriptors) {
for (final PropertyDescriptor descriptor : descriptors) {
orderedProperties.put(descriptor, null);
}
}
@ -1201,7 +1201,8 @@ public final class DtoFactory {
final PropertyDescriptor descriptor = entry.getKey();
// store the property descriptor
dto.getDescriptors().put(descriptor.getName(), createPropertyDescriptorDto(descriptor, controllerServiceNode.getProcessGroup().getIdentifier()));
final String groupId = controllerServiceNode.getProcessGroup() == null ? null : controllerServiceNode.getProcessGroup().getIdentifier();
dto.getDescriptors().put(descriptor.getName(), createPropertyDescriptorDto(descriptor, groupId));
// determine the property value - don't include sensitive properties
String propertyValue = entry.getValue();
@ -1254,7 +1255,7 @@ public final class DtoFactory {
propertyDescriptors = node.getControllerServiceImplementation().getPropertyDescriptors();
validationErrors = node.getValidationErrors();
processGroupId = node.getProcessGroup().getIdentifier();
processGroupId = node.getProcessGroup() == null ? null : node.getProcessGroup().getIdentifier();
} else if (component instanceof ReportingTaskNode) {
final ReportingTaskNode node = ((ReportingTaskNode) component);
dto.setState(node.getScheduledState().name());
@ -1264,20 +1265,20 @@ public final class DtoFactory {
propertyDescriptors = node.getReportingTask().getPropertyDescriptors();
validationErrors = node.getValidationErrors();
processGroupId = "root";
processGroupId = null;
}
if (propertyDescriptors != null && !propertyDescriptors.isEmpty()) {
final Map<PropertyDescriptor, String> sortedProperties = new TreeMap<>(new Comparator<PropertyDescriptor>() {
@Override
public int compare(PropertyDescriptor o1, PropertyDescriptor o2) {
public int compare(final PropertyDescriptor o1, final PropertyDescriptor o2) {
return Collator.getInstance(Locale.US).compare(o1.getName(), o2.getName());
}
});
sortedProperties.putAll(component.getProperties());
final Map<PropertyDescriptor, String> orderedProperties = new LinkedHashMap<>();
for (PropertyDescriptor descriptor : propertyDescriptors) {
for (final PropertyDescriptor descriptor : propertyDescriptors) {
orderedProperties.put(descriptor, null);
}
orderedProperties.putAll(sortedProperties);
@ -1717,7 +1718,7 @@ public final class DtoFactory {
dto.setComments(group.getComments());
dto.setName(group.getName());
ProcessGroup parentGroup = group.getParent();
final ProcessGroup parentGroup = group.getParent();
if (parentGroup != null) {
dto.setParentGroupId(parentGroup.getIdentifier());
}
@ -1886,7 +1887,7 @@ public final class DtoFactory {
// sort the relationships
Collections.sort(relationships, new Comparator<RelationshipDTO>() {
@Override
public int compare(RelationshipDTO r1, RelationshipDTO r2) {
public int compare(final RelationshipDTO r1, final RelationshipDTO r2) {
return Collator.getInstance(Locale.US).compare(r1.getName(), r2.getName());
}
});
@ -1923,7 +1924,7 @@ public final class DtoFactory {
// sort the bulletins
Collections.sort(bulletins, new Comparator<BulletinDTO>() {
@Override
public int compare(BulletinDTO bulletin1, BulletinDTO bulletin2) {
public int compare(final BulletinDTO bulletin1, final BulletinDTO bulletin2) {
if (bulletin1 == null && bulletin2 == null) {
return 0;
} else if (bulletin1 == null) {
@ -2260,7 +2261,7 @@ public final class DtoFactory {
// sort a copy of the properties
final Map<PropertyDescriptor, String> sortedProperties = new TreeMap<>(new Comparator<PropertyDescriptor>() {
@Override
public int compare(PropertyDescriptor o1, PropertyDescriptor o2) {
public int compare(final PropertyDescriptor o1, final PropertyDescriptor o2) {
return Collator.getInstance(Locale.US).compare(o1.getName(), o2.getName());
}
});
@ -2271,7 +2272,7 @@ public final class DtoFactory {
final Map<PropertyDescriptor, String> orderedProperties = new LinkedHashMap<>();
final List<PropertyDescriptor> descriptors = processor.getPropertyDescriptors();
if (descriptors != null && !descriptors.isEmpty()) {
for (PropertyDescriptor descriptor : descriptors) {
for (final PropertyDescriptor descriptor : descriptors) {
orderedProperties.put(descriptor, null);
}
}
@ -2798,7 +2799,7 @@ public final class DtoFactory {
* @param lastMod mod
* @return dto
*/
public RevisionDTO createRevisionDTO(FlowModification lastMod) {
public RevisionDTO createRevisionDTO(final FlowModification lastMod) {
final Revision revision = lastMod.getRevision();
// create the dto
@ -2817,7 +2818,7 @@ public final class DtoFactory {
return dto;
}
public NodeDTO createNodeDTO(NodeIdentifier nodeId, NodeConnectionStatus status, NodeHeartbeat nodeHeartbeat, List<NodeEvent> events, boolean primary) {
public NodeDTO createNodeDTO(final NodeIdentifier nodeId, final NodeConnectionStatus status, final NodeHeartbeat nodeHeartbeat, final List<NodeEvent> events, final boolean primary) {
final NodeDTO nodeDto = new NodeDTO();
// populate node dto
@ -2844,7 +2845,7 @@ public final class DtoFactory {
final List<NodeEvent> nodeEvents = new ArrayList<>(events);
Collections.sort(nodeEvents, new Comparator<NodeEvent>() {
@Override
public int compare(NodeEvent event1, NodeEvent event2) {
public int compare(final NodeEvent event1, final NodeEvent event2) {
return new Date(event2.getTimestamp()).compareTo(new Date(event1.getTimestamp()));
}
});
@ -2868,15 +2869,15 @@ public final class DtoFactory {
/* setters */
public void setControllerServiceProvider(ControllerServiceProvider controllerServiceProvider) {
public void setControllerServiceProvider(final ControllerServiceProvider controllerServiceProvider) {
this.controllerServiceProvider = controllerServiceProvider;
}
public void setAuthorizer(Authorizer authorizer) {
public void setAuthorizer(final Authorizer authorizer) {
this.authorizer = authorizer;
}
public void setEntityFactory(EntityFactory entityFactory) {
public void setEntityFactory(final EntityFactory entityFactory) {
this.entityFactory = entityFactory;
}

5
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/ControllerServiceDAO.java
View File

@ -51,11 +51,12 @@ public interface ControllerServiceDAO {
ControllerServiceNode getControllerService(String controllerServiceId);
/**
* Gets all of the controller services.
* Gets all of the controller services for the group with the given ID or all
* controller-level services if the group id is null
*
* @return The controller services
*/
Set<ControllerServiceNode> getControllerServices();
Set<ControllerServiceNode> getControllerServices(String groupId);
/**
* Updates the specified controller service.

66
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardControllerServiceDAO.java
View File

@ -16,31 +16,34 @@
*/
package org.apache.nifi.web.dao.impl;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.nifi.components.state.Scope;
import org.apache.nifi.components.state.StateMap;
import org.apache.nifi.controller.ConfiguredComponent;
import org.apache.nifi.controller.FlowController;
import org.apache.nifi.controller.ScheduledState;
import org.apache.nifi.controller.exception.ControllerServiceInstantiationException;
import org.apache.nifi.controller.exception.ValidationException;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.service.ControllerServiceProvider;
import org.apache.nifi.controller.service.ControllerServiceState;
import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.web.NiFiCoreException;
import org.apache.nifi.web.ResourceNotFoundException;
import org.apache.nifi.web.api.dto.ControllerServiceDTO;
import org.apache.nifi.web.dao.ComponentStateDAO;
import org.apache.nifi.web.dao.ControllerServiceDAO;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
public class StandardControllerServiceDAO extends ComponentDAO implements ControllerServiceDAO {
private ControllerServiceProvider serviceProvider;
private ComponentStateDAO componentStateDAO;
private FlowController flowController;
private ControllerServiceNode locateControllerService(final String controllerServiceId) {
// get the controller service
@ -71,6 +74,24 @@ public class StandardControllerServiceDAO extends ComponentDAO implements Contro
// perform the update
configureControllerService(controllerService, controllerServiceDTO);
final String groupId = controllerServiceDTO.getParentGroupId();
if (groupId == null) {
flowController.addRootControllerService(controllerService);
} else {
final ProcessGroup group;
if (groupId.equals(FlowController.ROOT_GROUP_ID_ALIAS)) {
group = flowController.getGroup(flowController.getRootGroupId());
} else {
group = flowController.getGroup(flowController.getRootGroupId()).findProcessGroup(groupId);
}
if (group == null) {
throw new ResourceNotFoundException(String.format("Unable to locate group with id '%s'.", groupId));
}
group.addControllerService(controllerService);
}
return controllerService;
} catch (final ControllerServiceInstantiationException csie) {
throw new NiFiCoreException(csie.getMessage(), csie);
@ -88,8 +109,17 @@ public class StandardControllerServiceDAO extends ComponentDAO implements Contro
}
@Override
public Set<ControllerServiceNode> getControllerServices() {
return serviceProvider.getAllControllerServices();
public Set<ControllerServiceNode> getControllerServices(final String groupId) {
if (groupId == null) {
return flowController.getRootControllerServices();
} else {
final ProcessGroup procGroup = flowController.getGroup(flowController.getRootGroupId()).findProcessGroup(groupId);
if (procGroup == null) {
throw new ResourceNotFoundException("Could not find Process Group with ID " + groupId);
}
return procGroup.getControllerServices(true);
}
}
@Override
@ -162,7 +192,7 @@ public class StandardControllerServiceDAO extends ComponentDAO implements Contro
}
@Override
public void verifyUpdateReferencingComponents(String controllerServiceId, ScheduledState scheduledState, ControllerServiceState controllerServiceState) {
public void verifyUpdateReferencingComponents(final String controllerServiceId, final ScheduledState scheduledState, final ControllerServiceState controllerServiceState) {
final ControllerServiceNode controllerService = locateControllerService(controllerServiceId);
if (controllerServiceState != null) {
@ -200,7 +230,7 @@ public class StandardControllerServiceDAO extends ComponentDAO implements Contro
controllerService.verifyCanDisable();
}
}
} catch (IllegalArgumentException iae) {
} catch (final IllegalArgumentException iae) {
throw new IllegalArgumentException("Controller Service state: Value must be one of [ENABLED, DISABLED]");
}
}
@ -255,35 +285,39 @@ public class StandardControllerServiceDAO extends ComponentDAO implements Contro
}
@Override
public void deleteControllerService(String controllerServiceId) {
public void deleteControllerService(final String controllerServiceId) {
final ControllerServiceNode controllerService = locateControllerService(controllerServiceId);
serviceProvider.removeControllerService(controllerService);
}
@Override
public StateMap getState(String controllerServiceId, Scope scope) {
public StateMap getState(final String controllerServiceId, final Scope scope) {
final ControllerServiceNode controllerService = locateControllerService(controllerServiceId);
return componentStateDAO.getState(controllerService, scope);
}
@Override
public void verifyClearState(String controllerServiceId) {
public void verifyClearState(final String controllerServiceId) {
final ControllerServiceNode controllerService = locateControllerService(controllerServiceId);
controllerService.verifyCanClearState();
}
@Override
public void clearState(String controllerServiceId) {
public void clearState(final String controllerServiceId) {
final ControllerServiceNode controllerService = locateControllerService(controllerServiceId);
componentStateDAO.clearState(controllerService);
}
/* setters */
public void setServiceProvider(ControllerServiceProvider serviceProvider) {
public void setServiceProvider(final ControllerServiceProvider serviceProvider) {
this.serviceProvider = serviceProvider;
}
public void setComponentStateDAO(ComponentStateDAO componentStateDAO) {
public void setComponentStateDAO(final ComponentStateDAO componentStateDAO) {
this.componentStateDAO = componentStateDAO;
}
public void setFlowController(final FlowController flowController) {
this.flowController = flowController;
}
}

3
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/resources/nifi-web-api-context.xml
View File

@ -91,6 +91,7 @@
<bean id="controllerServiceDAO" class="org.apache.nifi.web.dao.impl.StandardControllerServiceDAO">
<property name="serviceProvider" ref="controllerServiceProvider"/>
<property name="componentStateDAO" ref="componentStateDAO"/>
<property name="flowController" ref="flowController" />
</bean>
<bean id="reportingTaskDAO" class="org.apache.nifi.web.dao.impl.StandardReportingTaskDAO">
<property name="reportingTaskProvider" ref="reportingTaskProvider"/>
@ -150,7 +151,7 @@
<property name="clusterCoordinator" ref="clusterCoordinator" />
<property name="requestReplicator" ref="requestReplicator" />
<property name="auditService" ref="auditService"/>
<property name="controllerServiceLookup" ref="controllerServiceProvider"/>
<property name="controllerServiceProvider" ref="controllerServiceProvider"/>
<property name="reportingTaskProvider" ref="reportingTaskProvider"/>
<property name="authorizer" ref="authorizer"/>
</bean>