This is an automated email from the ASF dual-hosted git repository.
mattyb149 pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/support/nifi-1.x by this push:
new 6789389c61 NIFI-13755 Improved Controller Service Enabling Process
(#9273)
6789389c61 is described below
commit 6789389c61d446cecb69dfbdc13514dbf6fa8277
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 | 76 ++++++++++++----------
.../StandardControllerServiceProviderTest.java | 67 ++++++++++++++++++-
.../TestStandardControllerServiceProvider.java | 8 +--
3 files changed, 111 insertions(+), 40 deletions(-)
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
index 1844a9f10b..c9edd24ed7 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
@@ -210,39 +210,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
@@ -304,7 +314,7 @@ public class StandardControllerServiceProvider implements
ControllerServiceProvi
}
}
} catch (Exception e) {
- logger.error("Failed to enable " + controllerServiceNode, 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));
@@ -497,14 +507,14 @@ public class StandardControllerServiceProvider implements
ControllerServiceProvi
if (serviceNode == null) {
final ReportingTaskNode taskNode =
flowManager.getReportingTaskNode(componentId);
if (taskNode == null) {
- final ParameterProviderNode parameterProviderNode =
flowManager.getParameterProvider(componentId);
- if (parameterProviderNode == null) {
- final FlowRegistryClientNode flowRegistryClientNode =
flowManager.getFlowRegistryClient(componentId);
- if (flowRegistryClientNode == null) {
- throw new IllegalStateException("Could not find
any Processor, Reporting Task, Parameter Provider, or Controller Service with
identifier " + componentId);
+ final ParameterProviderNode parameterProviderNode =
flowManager.getParameterProvider(componentId);
+ if (parameterProviderNode == null) {
+ final FlowRegistryClientNode
flowRegistryClientNode = flowManager.getFlowRegistryClient(componentId);
+ if (flowRegistryClientNode == null) {
+ throw new IllegalStateException("Could not
find any Processor, Reporting Task, Parameter Provider, or Controller Service
with identifier " + componentId);
+ }
}
}
- }
// We have confirmed that the component is a reporting task or
parameter provider. We can only reference Controller Services
// that are scoped at the FlowController level in this case.
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java
index 1e06197de7..da819ca013 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java
@@ -39,9 +39,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 {
@@ -74,7 +80,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)
@@ -87,8 +93,6 @@ public class StandardControllerServiceProviderTest {
.stateManagerProvider(Mockito.mock(StateManagerProvider.class))
.extensionManager(extensionManager)
.buildControllerService();
-
- return serviceNode;
}
@Test
@@ -112,4 +116,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-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 027310409f..2ca70bbeb5 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
@@ -578,10 +578,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());
