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 <[email protected]>
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 <[email protected]>
---
.../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());
