You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ex...@apache.org on 2024/10/18 18:05:56 UTC
(nifi) branch main updated: NIFI-13755 Improved Controller Service Enabling Process (#9273)
This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 1aabc615ce NIFI-13755 Improved Controller Service Enabling Process (#9273)
1aabc615ce is described below
commit 1aabc615ce923a07281bcb833e247ad3b51bcf42
Author: jrsteinebrey <jr...@gmail.com>
AuthorDate: Fri Oct 18 14:05:50 2024 -0400
NIFI-13755 Improved Controller Service Enabling Process (#9273)
- Updated Standard Controller Service Provider to enable requested Controller Services that do not depend on a disabled Controller Service, instead of failing to enable the entire set of requested Controller Services
- Updated enabling process improves behavior when restarting NiFi and attempting to enable large numbers of Controller Services, some of which depend on disabled Services
Signed-off-by: David Handermann <ex...@apache.org>
---
.../service/StandardControllerServiceProvider.java | 62 +++++++++++---------
.../StandardControllerServiceProviderTest.java | 67 +++++++++++++++++++++-
.../TestStandardControllerServiceProvider.java | 8 +--
3 files changed, 104 insertions(+), 33 deletions(-)
diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
index cf4a53d3e2..0629458332 100644
--- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
+++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
@@ -252,39 +252,49 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi
}
@Override
- public void enableControllerServices(final Collection<ControllerServiceNode> serviceNodes) {
- boolean shouldStart = true;
-
- Iterator<ControllerServiceNode> serviceIter = serviceNodes.iterator();
- while (serviceIter.hasNext() && shouldStart) {
- ControllerServiceNode controllerServiceNode = serviceIter.next();
- List<ControllerServiceNode> requiredServices = controllerServiceNode.getRequiredControllerServices();
- for (ControllerServiceNode requiredService : requiredServices) {
- if (!requiredService.isActive() && !serviceNodes.contains(requiredService)) {
- shouldStart = false;
- logger.debug("Will not start {} because required service {} is not active and is not part of the collection of things to start", serviceNodes, requiredService);
+ public void enableControllerServices(final Collection<ControllerServiceNode> serviceNodesIn) {
+ Collection<ControllerServiceNode> serviceNodes = new HashSet<>(serviceNodesIn);
+ for (ControllerServiceNode controllerServiceNode : removeControllerServicesWithUnavailableRequirements(serviceNodes)) {
+ try {
+ final Future<Void> future = enableControllerServiceAndDependencies(controllerServiceNode);
+
+ future.get(30, TimeUnit.SECONDS);
+ logger.debug("Successfully enabled {}; service state = {}", controllerServiceNode, controllerServiceNode.getState());
+ } catch (final ControllerServiceNotValidException csnve) {
+ logger.warn("Failed to enable service {} because it is not currently valid", controllerServiceNode);
+ } catch (Exception e) {
+ logger.error("Failed to enable {}", controllerServiceNode, e);
+ if (this.bulletinRepo != null) {
+ this.bulletinRepo.addBulletin(BulletinFactory.createBulletin("Controller Service",
+ Severity.ERROR.name(), "Could not start " + controllerServiceNode + " due to " + e));
}
}
}
+ }
- if (shouldStart) {
- for (ControllerServiceNode controllerServiceNode : serviceNodes) {
- try {
- final Future<Void> future = enableControllerServiceAndDependencies(controllerServiceNode);
-
- future.get(30, TimeUnit.SECONDS);
- logger.debug("Successfully enabled {}; service state = {}", controllerServiceNode, controllerServiceNode.getState());
- } catch (final ControllerServiceNotValidException csnve) {
- logger.warn("Failed to enable service {} because it is not currently valid", controllerServiceNode);
- } catch (Exception e) {
- logger.error("Failed to enable {}", controllerServiceNode, e);
- if (this.bulletinRepo != null) {
- this.bulletinRepo.addBulletin(BulletinFactory.createBulletin("Controller Service",
- Severity.ERROR.name(), "Could not start " + controllerServiceNode + " due to " + e));
+ private Collection<ControllerServiceNode> removeControllerServicesWithUnavailableRequirements(final Collection<ControllerServiceNode> serviceNodes) {
+ boolean recheckNeeded;
+ do {
+ recheckNeeded = false;
+ for (Iterator<ControllerServiceNode> iter = serviceNodes.iterator(); iter.hasNext();) {
+ boolean skipStarting = false;
+ final ControllerServiceNode serviceNode = iter.next();
+ final List<ControllerServiceNode> requiredServices = serviceNode.getRequiredControllerServices();
+ for (ControllerServiceNode requiredService : requiredServices) {
+ if (!requiredService.isActive() && !serviceNodes.contains(requiredService)) {
+ skipStarting = true;
+ logger.error("Will not start {} because its required service {} is not active and is not part of the collection of things to start", serviceNode, requiredService);
}
}
+ if (skipStarting) {
+ // If any service was removed, then recheck all remaining services because the removed one might be required by another service in the list.
+ recheckNeeded = true;
+ iter.remove();
+ }
}
- }
+ } while (recheckNeeded);
+
+ return serviceNodes;
}
@Override
diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java
index b1ed391694..2153c63043 100644
--- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java
+++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java
@@ -37,9 +37,15 @@ import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
public class StandardControllerServiceProviderTest {
@@ -69,7 +75,7 @@ public class StandardControllerServiceProviderTest {
}
private ControllerServiceNode createControllerService(final String type, final String id, final BundleCoordinate bundleCoordinate, final ControllerServiceProvider serviceProvider) {
- final ControllerServiceNode serviceNode = new ExtensionBuilder()
+ return new ExtensionBuilder()
.identifier(id)
.type(type)
.bundleCoordinate(bundleCoordinate)
@@ -81,8 +87,6 @@ public class StandardControllerServiceProviderTest {
.stateManagerProvider(Mockito.mock(StateManagerProvider.class))
.extensionManager(extensionManager)
.buildControllerService();
-
- return serviceNode;
}
@Test
@@ -106,4 +110,61 @@ public class StandardControllerServiceProviderTest {
public void testCallImplementationInitialized() throws InitializationException {
implementation.initialize(null);
}
+
+ private ControllerServiceNode populateControllerService(ControllerServiceNode requiredService) { // Collection<ControllerServiceNode> serviceNodes) {
+ ControllerServiceNode controllerServiceNode = mock(ControllerServiceNode.class);
+ ArrayList<ControllerServiceNode> requiredServices = new ArrayList<>();
+ if (requiredService != null) {
+ requiredServices.add(requiredService);
+ }
+ when(controllerServiceNode.getRequiredControllerServices()).thenReturn(requiredServices);
+ return controllerServiceNode;
+ }
+
+ @Test
+ public void testEnableControllerServicesAllAreEnabled() {
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ future.complete(null);
+
+ ProcessScheduler scheduler = Mockito.mock(ProcessScheduler.class);
+ when(scheduler.enableControllerService(any())).thenReturn(future);
+ ControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, Mockito.mock(FlowManager.class), Mockito.mock(ExtensionManager.class));
+
+ final ArrayList<ControllerServiceNode> serviceNodes = new ArrayList<>();
+ serviceNodes.add(populateControllerService(null));
+ serviceNodes.add(populateControllerService(null));
+ provider.enableControllerServices(serviceNodes);
+ verify(scheduler).enableControllerService(serviceNodes.get(0));
+ verify(scheduler).enableControllerService(serviceNodes.get(1));
+ }
+
+ @Test
+ public void testEnableControllerServicesSomeAreEnabled() {
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ future.complete(null);
+
+ ProcessScheduler scheduler = Mockito.mock(ProcessScheduler.class);
+ when(scheduler.enableControllerService(any())).thenReturn(future);
+ ControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, Mockito.mock(FlowManager.class), Mockito.mock(ExtensionManager.class));
+
+ final ArrayList<ControllerServiceNode> serviceNodes = new ArrayList<>();
+ ControllerServiceNode disabledController = populateControllerService(null);
+ // Do not start because disabledController is not in the serviceNodes (list of services to start)
+ serviceNodes.add(populateControllerService(disabledController));
+ // Start this service because it has no required services
+ serviceNodes.add(populateControllerService(null));
+ // Do not start because its required service is not started because the required service
+ // depends on disabledController which is not in the serviceNodes (list of services to start)
+ serviceNodes.add(populateControllerService(serviceNodes.get(0)));
+ // Do not start because its required service depends on disabledController through 2 other levels of services
+ serviceNodes.add(populateControllerService(serviceNodes.get(2)));
+ // Start this service because it has a required service which is in the list of services to start
+ serviceNodes.add(populateControllerService(serviceNodes.get(1)));
+ provider.enableControllerServices(serviceNodes);
+ verify(scheduler, Mockito.times(0)).enableControllerService(serviceNodes.get(0));
+ verify(scheduler, Mockito.times(2)).enableControllerService(serviceNodes.get(1));
+ verify(scheduler, Mockito.times(0)).enableControllerService(serviceNodes.get(2));
+ verify(scheduler, Mockito.times(0)).enableControllerService(serviceNodes.get(3));
+ verify(scheduler, Mockito.times(1)).enableControllerService(serviceNodes.get(4));
+ }
}
diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java
index f0f27e787a..05e54d3a07 100644
--- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java
+++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java
@@ -576,10 +576,10 @@ public class TestStandardControllerServiceProvider {
allBut6.stream().forEach(ControllerServiceNode::performValidation);
provider.enableControllerServices(allBut6);
- assertFalse(serviceNode1.isActive());
- assertFalse(serviceNode2.isActive());
- assertFalse(serviceNode3.isActive());
- assertFalse(serviceNode4.isActive());
+ assertTrue(serviceNode1.isActive());
+ assertTrue(serviceNode2.isActive());
+ assertTrue(serviceNode3.isActive());
+ assertTrue(serviceNode4.isActive());
assertFalse(serviceNode5.isActive());
assertFalse(serviceNode6.isActive());