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

NIFI-2160 fixed service startup ordering (MASTER) 

This closes #606
This commit is contained in:
Oleg Zhurakousky 2016-07-05 20:08:34 -04:00
parent 6edfa634d4
commit 2ed1ab7630
3 changed files with 82 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -22,8 +22,8 @@ import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@ -170,7 +170,7 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i
@Override
public List<ControllerServiceNode> getRequiredControllerServices() {
List<ControllerServiceNode> requiredServices = new ArrayList<>();
Set<ControllerServiceNode> requiredServices = new HashSet<>();
for (Entry<PropertyDescriptor, String> pEntry : this.getProperties().entrySet()) {
PropertyDescriptor descriptor = pEntry.getKey();
if (descriptor.getControllerServiceDefinition() != null && descriptor.isRequired()) {
@ -179,7 +179,7 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i
requiredServices.addAll(rNode.getRequiredControllerServices());
}
}
return requiredServices;
return new ArrayList<>(requiredServices);
}

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

@ -25,7 +25,6 @@ import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
@ -55,7 +54,6 @@ import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.nar.NarCloseable;
import org.apache.nifi.processor.SimpleProcessLogger;
import org.apache.nifi.processor.StandardValidationContextFactory;
import org.apache.nifi.reporting.BulletinRepository;
import org.apache.nifi.reporting.Severity;
import org.apache.nifi.util.ObjectHolder;
@ -385,17 +383,11 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi
}
if (shouldStart) {
List<ControllerServiceNode> services = new ArrayList<>(serviceNodes);
Collections.sort(services, new Comparator<ControllerServiceNode>() {
@Override
public int compare(ControllerServiceNode s1, ControllerServiceNode s2) {
return s2.getRequiredControllerServices().contains(s1) ? -1 : 1;
}
});
for (ControllerServiceNode controllerServiceNode : services) {
for (ControllerServiceNode controllerServiceNode : serviceNodes) {
try {
this.enableControllerService(controllerServiceNode);
if (!controllerServiceNode.isActive()) {
this.enableControllerServiceDependenciesFirst(controllerServiceNode);
}
} catch (Exception e) {
logger.error("Failed to enable " + controllerServiceNode + " due to " + e);
if (this.bulletinRepo != null) {
@ -407,6 +399,18 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi
}
}
private void enableControllerServiceDependenciesFirst(ControllerServiceNode serviceNode) {
for (ControllerServiceNode depNode : serviceNode.getRequiredControllerServices()) {
if (!depNode.isActive()) {
this.enableControllerServiceDependenciesFirst(depNode);
}
}
if (logger.isDebugEnabled()) {
logger.debug("Enabling " + serviceNode);
}
this.enableControllerService(serviceNode);
}
static List<List<ControllerServiceNode>> determineEnablingOrder(final Map<String, ControllerServiceNode> serviceNodeMap) {
final List<List<ControllerServiceNode>> orderedNodeLists = new ArrayList<>();

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

@ -398,32 +398,73 @@ public class TestStandardControllerServiceProvider {
ProcessGroup procGroup = new MockProcessGroup();
Mockito.when(controller.getGroup(Mockito.anyString())).thenReturn(procGroup);
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 A = provider.createControllerService(ServiceA.class.getName(), "A", false);
ControllerServiceNode B = provider.createControllerService(ServiceA.class.getName(), "B", false);
ControllerServiceNode C = provider.createControllerService(ServiceA.class.getName(), "C", false);
ControllerServiceNode D = provider.createControllerService(ServiceB.class.getName(), "D", false);
ControllerServiceNode E = provider.createControllerService(ServiceA.class.getName(), "E", false);
procGroup.addControllerService(serviceNode1);
procGroup.addControllerService(serviceNode2);
procGroup.addControllerService(serviceNode3);
procGroup.addControllerService(serviceNode4);
procGroup.addControllerService(serviceNode5);
procGroup.addControllerService(A);
procGroup.addControllerService(B);
procGroup.addControllerService(C);
procGroup.addControllerService(D);
procGroup.addControllerService(E);
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");
A.setProperty(ServiceA.OTHER_SERVICE.getName(), "B");
B.setProperty(ServiceA.OTHER_SERVICE.getName(), "D");
C.setProperty(ServiceA.OTHER_SERVICE.getName(), "B");
C.setProperty(ServiceA.OTHER_SERVICE_2.getName(), "D");
E.setProperty(ServiceA.OTHER_SERVICE.getName(), "A");
provider.enableControllerServices(
Arrays.asList(new ControllerServiceNode[] { serviceNode1, serviceNode2, serviceNode3, serviceNode4, serviceNode5}));
provider.enableControllerServices(Arrays.asList(new ControllerServiceNode[] { A, B, C, D, E }));
assertTrue(serviceNode1.isActive());
assertTrue(serviceNode2.isActive());
assertTrue(serviceNode3.isActive());
assertTrue(serviceNode4.isActive());
assertTrue(serviceNode5.isActive());
assertTrue(A.isActive());
assertTrue(B.isActive());
assertTrue(C.isActive());
assertTrue(D.isActive());
assertTrue(E.isActive());
}
/**
* This test is similar to the above, but different combination of service
* dependencies
*
*/
@Test
public void validateEnableServices2() {
StandardProcessScheduler scheduler = createScheduler();
FlowController controller = Mockito.mock(FlowController.class);
StandardControllerServiceProvider provider = new StandardControllerServiceProvider(controller, scheduler, null,
stateManagerProvider);
ProcessGroup procGroup = new MockProcessGroup();
Mockito.when(controller.getGroup(Mockito.anyString())).thenReturn(procGroup);
ControllerServiceNode A = provider.createControllerService(ServiceC.class.getName(), "A", false);
ControllerServiceNode B = provider.createControllerService(ServiceA.class.getName(), "B", false);
ControllerServiceNode C = provider.createControllerService(ServiceB.class.getName(), "C", false);
ControllerServiceNode D = provider.createControllerService(ServiceA.class.getName(), "D", false);
ControllerServiceNode F = provider.createControllerService(ServiceA.class.getName(), "F", false);
procGroup.addControllerService(A);
procGroup.addControllerService(B);
procGroup.addControllerService(C);
procGroup.addControllerService(D);
procGroup.addControllerService(F);
A.setProperty(ServiceC.REQ_SERVICE_1.getName(), "B");
A.setProperty(ServiceC.REQ_SERVICE_2.getName(), "D");
B.setProperty(ServiceA.OTHER_SERVICE.getName(), "C");
F.setProperty(ServiceA.OTHER_SERVICE.getName(), "D");
D.setProperty(ServiceA.OTHER_SERVICE.getName(), "C");
provider.enableControllerServices(Arrays.asList(new ControllerServiceNode[] { C, F, A, B, D }));
assertTrue(A.isActive());
assertTrue(B.isActive());
assertTrue(C.isActive());
assertTrue(D.isActive());
assertTrue(F.isActive());
}
@Test