Repository: nifi Updated Branches: refs/heads/0.x f0c9fa631 -> c6109d449
NIFI-2032 port for 0.x (+1 squashed commit) Squashed commits: [a9b32c6] Fixed checkstyle issues in TestStandardControllerServiceProvider. This closes #554. Signed-off-by: Andy LoPresto <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/c6109d44 Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/c6109d44 Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/c6109d44 Branch: refs/heads/0.x Commit: c6109d4494417111bcc40701873eed652f3d34c3 Parents: f0c9fa6 Author: Oleg Zhurakousky <[email protected]> Authored: Fri Jun 17 12:08:26 2016 -0400 Committer: Andy LoPresto <[email protected]> Committed: Tue Jun 21 17:03:48 2016 -0700 ---------------------------------------------------------------------- .../service/ControllerServiceNode.java | 11 ++ .../service/StandardControllerServiceNode.java | 19 ++++ .../StandardControllerServiceProvider.java | 107 +++++-------------- .../TestStandardControllerServiceProvider.java | 73 ++++++++++++- .../nifi/controller/service/mock/ServiceC.java | 55 ++++++++++ 5 files changed, 186 insertions(+), 79 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/c6109d44/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java index e91ba9a..6376c62 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java @@ -16,6 +16,7 @@ */ package org.apache.nifi.controller.service; +import java.util.List; import java.util.Set; import java.util.concurrent.ScheduledExecutorService; @@ -38,6 +39,16 @@ public interface ControllerServiceNode extends ConfiguredComponent { ControllerService getProxiedControllerService(); /** + * Returns the list of services that are required to be enabled before this + * service is enabled. The returned list is flattened and contains both + * immediate and transient dependencies. + * + * @return list of services required to be enabled before this service is + * enabled + */ + List<ControllerServiceNode> getRequiredControllerServices(); + + /** * <p> * Returns the actual implementation of the Controller Service that this ControllerServiceNode * encapsulates. This direct implementation should <strong>NEVER</strong> be passed to another http://git-wip-us.apache.org/repos/asf/nifi/blob/c6109d44/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java index 83c35ad..2deb4ed 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java @@ -17,9 +17,12 @@ package org.apache.nifi.controller.service; import java.lang.reflect.InvocationTargetException; +import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.HashSet; +import java.util.List; +import java.util.Map.Entry; import java.util.Set; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -31,6 +34,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock; import org.apache.nifi.annotation.lifecycle.OnDisabled; import org.apache.nifi.annotation.lifecycle.OnEnabled; +import org.apache.nifi.components.PropertyDescriptor; import org.apache.nifi.components.ValidationResult; import org.apache.nifi.controller.AbstractConfiguredComponent; import org.apache.nifi.controller.ConfigurationContext; @@ -110,6 +114,21 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i } @Override + public List<ControllerServiceNode> getRequiredControllerServices() { + List<ControllerServiceNode> requiredServices = new ArrayList<>(); + for (Entry<PropertyDescriptor, String> pEntry : this.getProperties().entrySet()) { + PropertyDescriptor descriptor = pEntry.getKey(); + if (descriptor.getControllerServiceDefinition() != null && descriptor.isRequired()) { + ControllerServiceNode rNode = this.serviceProvider.getControllerServiceNode(pEntry.getValue()); + requiredServices.add(rNode); + requiredServices.addAll(rNode.getRequiredControllerServices()); + } + } + return requiredServices; + } + + + @Override public void removeReference(final ConfiguredComponent referencingComponent) { writeLock.lock(); try { http://git-wip-us.apache.org/repos/asf/nifi/blob/c6109d44/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java index d0ad127..4b18751 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java @@ -25,15 +25,14 @@ import java.lang.reflect.Proxy; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; -import java.util.HashMap; +import java.util.Comparator; import java.util.HashSet; +import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; import org.apache.nifi.annotation.lifecycle.OnAdded; import org.apache.nifi.annotation.lifecycle.OnRemoved; @@ -300,88 +299,40 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi @Override public void enableControllerServices(final Collection<ControllerServiceNode> serviceNodes) { - final Set<ControllerServiceNode> servicesToEnable = new HashSet<>(); - // Ensure that all nodes are already disabled - for (final ControllerServiceNode serviceNode : serviceNodes) { - final ControllerServiceState curState = serviceNode.getState(); - if (ControllerServiceState.DISABLED.equals(curState)) { - servicesToEnable.add(serviceNode); - } else { - logger.warn("Cannot enable {} because it is not disabled; current state is {}", serviceNode, curState); + boolean shouldStart = true; + + Iterator<ControllerServiceNode> serviceIter = serviceNodes.iterator(); + while (serviceIter.hasNext() && shouldStart) { + ControllerServiceNode controllerServiceNode = serviceIter.next(); + List<ControllerServiceNode> requiredServices = ((StandardControllerServiceNode) controllerServiceNode).getRequiredControllerServices(); + for (ControllerServiceNode requiredService : requiredServices) { + if (!requiredService.isActive() && !serviceNodes.contains(requiredService)) { + shouldStart = false; + } } } - // determine the order to load the services. We have to ensure that if service A references service B, then B - // is enabled first, and so on. - final Map<String, ControllerServiceNode> idToNodeMap = new HashMap<>(); - for (final ControllerServiceNode node : servicesToEnable) { - idToNodeMap.put(node.getIdentifier(), node); - } - - // We can have many Controller Services dependent on one another. We can have many of these - // disparate lists of Controller Services that are dependent on one another. We refer to each - // of these as a branch. - final List<List<ControllerServiceNode>> branches = determineEnablingOrder(idToNodeMap); - - if (branches.isEmpty()) { - logger.info("No Controller Services to enable"); - return; - } else { - logger.info("Will enable {} Controller Services", servicesToEnable.size()); - } - - final Set<ControllerServiceNode> enabledNodes = Collections.synchronizedSet(new HashSet<ControllerServiceNode>()); - final ExecutorService executor = Executors.newFixedThreadPool(Math.min(10, branches.size())); - for (final List<ControllerServiceNode> branch : branches) { - final Runnable enableBranchRunnable = new Runnable() { + if (shouldStart) { + List<ControllerServiceNode> services = new ArrayList<>(serviceNodes); + Collections.sort(services, new Comparator<ControllerServiceNode>() { @Override - public void run() { - logger.debug("Enabling Controller Service Branch {}", branch); - - for (final ControllerServiceNode serviceNode : branch) { - try { - if (!enabledNodes.contains(serviceNode)) { - enabledNodes.add(serviceNode); - - logger.info("Enabling {}", serviceNode); - try { - processScheduler.enableControllerService(serviceNode); - } catch (final Exception e) { - logger.error("Failed to enable " + serviceNode + " due to " + e); - if (logger.isDebugEnabled()) { - logger.error("", e); - } - - if (bulletinRepo != null) { - bulletinRepo.addBulletin(BulletinFactory.createBulletin( - "Controller Service", Severity.ERROR.name(), "Could not start " + serviceNode + " due to " + e)); - } - } - } - - // wait for service to finish enabling. - while (ControllerServiceState.ENABLING.equals(serviceNode.getState())) { - try { - Thread.sleep(100L); - } catch (final InterruptedException ie) { - } - } - - logger.info("State for {} is now {}", serviceNode, serviceNode.getState()); - } catch (final Exception e) { - logger.error("Failed to enable {} due to {}", serviceNode, e.toString()); - if (logger.isDebugEnabled()) { - logger.error("", e); - } - } + public int compare(ControllerServiceNode s1, ControllerServiceNode s2) { + return s2.getRequiredControllerServices().contains(s1) ? -1 : 1; + } + }); + + for (ControllerServiceNode controllerServiceNode : services) { + try { + this.enableControllerService(controllerServiceNode); + } catch (Exception e) { + logger.error("Failed to enable " + controllerServiceNode + " due to " + e); + if (this.bulletinRepo != null) { + this.bulletinRepo.addBulletin(BulletinFactory.createBulletin("Controller Service", + Severity.ERROR.name(), "Could not start " + controllerServiceNode + " due to " + e)); } } - }; - - executor.submit(enableBranchRunnable); + } } - - executor.shutdown(); } static List<List<ControllerServiceNode>> determineEnablingOrder(final Map<String, ControllerServiceNode> serviceNodeMap) { http://git-wip-us.apache.org/repos/asf/nifi/blob/c6109d44/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java index a11610d..8041aef 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java @@ -17,16 +17,17 @@ package org.apache.nifi.controller.service; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import java.beans.PropertyDescriptor; +import java.util.Arrays; import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; import java.util.UUID; - import org.apache.nifi.components.state.StateManager; import org.apache.nifi.components.state.StateManagerProvider; import org.apache.nifi.controller.Heartbeater; @@ -38,6 +39,7 @@ import org.apache.nifi.controller.scheduling.StandardProcessScheduler; import org.apache.nifi.controller.service.mock.DummyProcessor; import org.apache.nifi.controller.service.mock.ServiceA; import org.apache.nifi.controller.service.mock.ServiceB; +import org.apache.nifi.controller.service.mock.ServiceC; import org.apache.nifi.groups.ProcessGroup; import org.apache.nifi.groups.StandardProcessGroup; import org.apache.nifi.processor.StandardValidationContextFactory; @@ -363,4 +365,73 @@ public class TestStandardControllerServiceProvider { provider.unscheduleReferencingComponents(serviceNode); assertEquals(ScheduledState.STOPPED, procNode.getScheduledState()); } + + @Test + public void validateEnableServices() { + StandardProcessScheduler scheduler = createScheduler(); + StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider, null); + + ControllerServiceNode serviceNode1 = provider.createControllerService(ServiceA.class.getName(), "1", false); + ControllerServiceNode serviceNode2 = provider.createControllerService(ServiceA.class.getName(), "2", false); + ControllerServiceNode serviceNode3 = provider.createControllerService(ServiceA.class.getName(), "3", false); + ControllerServiceNode serviceNode4 = provider.createControllerService(ServiceB.class.getName(), "4", false); + ControllerServiceNode serviceNode5 = provider.createControllerService(ServiceA.class.getName(), "5", false); + + serviceNode1.setProperty(ServiceA.OTHER_SERVICE.getName(), "2"); + serviceNode2.setProperty(ServiceA.OTHER_SERVICE.getName(), "4"); + serviceNode3.setProperty(ServiceA.OTHER_SERVICE.getName(), "2"); + serviceNode3.setProperty(ServiceA.OTHER_SERVICE_2.getName(), "4"); + serviceNode5.setProperty(ServiceA.OTHER_SERVICE.getName(), "1"); + + provider.enableControllerServices( + Arrays.asList(new ControllerServiceNode[] { serviceNode1, serviceNode2, serviceNode3, serviceNode4, serviceNode5})); + + assertTrue(serviceNode1.isActive()); + assertTrue(serviceNode2.isActive()); + assertTrue(serviceNode3.isActive()); + assertTrue(serviceNode4.isActive()); + assertTrue(serviceNode5.isActive()); + } + + @Test + public void validateEnableServicesWithDisabledMissingService() { + StandardProcessScheduler scheduler = createScheduler(); + StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider, null); + + ControllerServiceNode serviceNode1 = provider.createControllerService(ServiceA.class.getName(), "1", false); + ControllerServiceNode serviceNode2 = provider.createControllerService(ServiceA.class.getName(), "2", false); + ControllerServiceNode serviceNode3 = provider.createControllerService(ServiceA.class.getName(), "3", false); + ControllerServiceNode serviceNode4 = provider.createControllerService(ServiceB.class.getName(), "4", false); + ControllerServiceNode serviceNode5 = provider.createControllerService(ServiceA.class.getName(), "5", false); + ControllerServiceNode serviceNode6 = provider.createControllerService(ServiceB.class.getName(), "6", false); + ControllerServiceNode serviceNode7 = provider.createControllerService(ServiceC.class.getName(), "7", false); + + serviceNode1.setProperty(ServiceA.OTHER_SERVICE.getName(), "2"); + serviceNode2.setProperty(ServiceA.OTHER_SERVICE.getName(), "4"); + serviceNode3.setProperty(ServiceA.OTHER_SERVICE.getName(), "2"); + serviceNode3.setProperty(ServiceA.OTHER_SERVICE_2.getName(), "4"); + serviceNode5.setProperty(ServiceA.OTHER_SERVICE.getName(), "6"); + serviceNode7.setProperty(ServiceC.REQ_SERVICE_1.getName(), "2"); + serviceNode7.setProperty(ServiceC.REQ_SERVICE_2.getName(), "3"); + + provider.enableControllerServices(Arrays.asList( + new ControllerServiceNode[] { serviceNode1, serviceNode2, serviceNode3, serviceNode4, serviceNode5, serviceNode7})); + assertFalse(serviceNode1.isActive()); + assertFalse(serviceNode2.isActive()); + assertFalse(serviceNode3.isActive()); + assertFalse(serviceNode4.isActive()); + assertFalse(serviceNode5.isActive()); + assertFalse(serviceNode6.isActive()); + + provider.enableControllerService(serviceNode6); + provider.enableControllerServices(Arrays.asList( + new ControllerServiceNode[] { serviceNode1, serviceNode2, serviceNode3, serviceNode4, serviceNode5 })); + + assertTrue(serviceNode1.isActive()); + assertTrue(serviceNode2.isActive()); + assertTrue(serviceNode3.isActive()); + assertTrue(serviceNode4.isActive()); + assertTrue(serviceNode5.isActive()); + assertTrue(serviceNode6.isActive()); + } } http://git-wip-us.apache.org/repos/asf/nifi/blob/c6109d44/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/mock/ServiceC.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/mock/ServiceC.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/mock/ServiceC.java new file mode 100644 index 0000000..8c46bb6 --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/mock/ServiceC.java @@ -0,0 +1,55 @@ +/* + * 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.controller.service.mock; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.controller.AbstractControllerService; +import org.apache.nifi.controller.ControllerService; + +public class ServiceC extends AbstractControllerService { + + public static final PropertyDescriptor REQ_SERVICE_1 = new PropertyDescriptor.Builder() + .name("S1") + .identifiesControllerService(ControllerService.class) + .required(true) + .build(); + + public static final PropertyDescriptor REQ_SERVICE_2 = new PropertyDescriptor.Builder() + .name("S2") + .identifiesControllerService(ControllerService.class) + .required(true) + .build(); + + public static final PropertyDescriptor OPT_SERVICE = new PropertyDescriptor.Builder() + .name("S3") + .identifiesControllerService(ControllerService.class) + .required(false) + .build(); + + @Override + protected List<PropertyDescriptor> getSupportedPropertyDescriptors() { + final List<PropertyDescriptor> descriptors = new ArrayList<>(); + descriptors.add(REQ_SERVICE_1); + descriptors.add(REQ_SERVICE_2); + descriptors.add(OPT_SERVICE); + return descriptors; + } + +}
