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 46fc537ec8a NIFI-16071 Enable newly added Controller Services on
versioned flow upgrade (#11390)
46fc537ec8a is described below
commit 46fc537ec8ad6e5112a05753ffe4110f133e1c90
Author: Pierre Villard <[email protected]>
AuthorDate: Fri Jul 3 03:01:04 2026 +0200
NIFI-16071 Enable newly added Controller Services on versioned flow upgrade
(#11390)
Signed-off-by: David Handermann <[email protected]>
---
.../StandardVersionedComponentSynchronizer.java | 10 +++
.../RetainExistingStateComponentScheduler.java | 8 ++
.../RetainExistingStateComponentSchedulerTest.java | 24 ++++++
.../org/apache/nifi/groups/ComponentScheduler.java | 18 ++++
.../processors/tests/system/CountFlowFiles.java | 11 ++-
.../tests/system/registry/RegistryClientIT.java | 99 ++++++++++++++++++++++
6 files changed, 169 insertions(+), 1 deletion(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
index a43aca79a8b..d8785cd9a2a 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
@@ -778,6 +778,16 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
context.getComponentScheduler().enableControllerServicesAsync(Collections.singleton(service));
}
});
+
+ // Controller Services are always persisted in a versioned flow as
DISABLED, so newly added services never appear
+ // in the "proposed state is ENABLED" set above. Let the
ComponentScheduler decide whether to enable them: on a
+ // versioned-flow upgrade of an active Process Group,
RetainExistingStateComponentScheduler enables them (mirroring
+ // how newly added processors are started), while startup/restore
schedulers treat this as a no-op so that a service
+ // that was not previously enabled remains disabled.
+ final Set<ControllerServiceNode> addedServices = new
HashSet<>(servicesAdded.values());
+ addedServices.removeAll(toEnable);
+ addedServices.forEach(ComponentNode::performValidation);
+
context.getComponentScheduler().enableAddedControllerServicesAsync(addedServices);
}
private void removeMissingConnections(final ProcessGroup group, final
VersionedProcessGroup proposed, final Map<String, Connection>
connectionsByVersionedId) {
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/RetainExistingStateComponentScheduler.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/RetainExistingStateComponentScheduler.java
index 393b9866bdf..ffcb92a84ee 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/RetainExistingStateComponentScheduler.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/RetainExistingStateComponentScheduler.java
@@ -158,6 +158,14 @@ public class RetainExistingStateComponentScheduler
implements ComponentScheduler
delegate.enableControllerServicesAsync(toEnable);
}
+ @Override
+ public void enableAddedControllerServicesAsync(final
Collection<ControllerServiceNode> controllerServices) {
+ // These services are newly added by the flow update (they were not
present before). Apply the same policy used
+ // for other newly added components: enable them only if the Process
Group is active. enableControllerServicesAsync
+ // already implements this gate for services with no previous state,
so delegate to it.
+ enableControllerServicesAsync(controllerServices);
+ }
+
@Override
public void disableControllerServicesAsync(final
Collection<ControllerServiceNode> controllerServices) {
delegate.disableControllerServicesAsync(controllerServices);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/RetainExistingStateComponentSchedulerTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/RetainExistingStateComponentSchedulerTest.java
index 8f5add62a79..5456e7c84f1 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/RetainExistingStateComponentSchedulerTest.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/RetainExistingStateComponentSchedulerTest.java
@@ -175,6 +175,30 @@ class RetainExistingStateComponentSchedulerTest {
verify(delegate).enableControllerServicesAsync(Collections.emptySet());
}
+ @Test
+ void testAddedServiceEnabledWhenProcessGroupActive() {
+ final ProcessorNode runningProcessor =
createMockProcessor("running-proc", ScheduledState.RUNNING);
+ final ProcessGroup group =
createProcessGroup(Set.of(runningProcessor), Collections.emptySet(),
Collections.emptySet());
+ final RetainExistingStateComponentScheduler scheduler = new
RetainExistingStateComponentScheduler(group, delegate);
+ assertTrue(scheduler.isProcessGroupActive());
+
+ // A service newly added by the update, referenced only by an existing
processor, must be enabled when the group is active.
+ final ControllerServiceNode addedService =
createMockService("added-svc", ControllerServiceState.DISABLED);
+ scheduler.enableAddedControllerServicesAsync(List.of(addedService));
+ verify(delegate).enableControllerServicesAsync(Set.of(addedService));
+ }
+
+ @Test
+ void testAddedServiceNotEnabledWhenProcessGroupInactive() {
+ final ProcessGroup group = createProcessGroup(Collections.emptySet(),
Collections.emptySet(), Collections.emptySet());
+ final RetainExistingStateComponentScheduler scheduler = new
RetainExistingStateComponentScheduler(group, delegate);
+ assertFalse(scheduler.isProcessGroupActive());
+
+ final ControllerServiceNode addedService =
createMockService("added-svc", ControllerServiceState.DISABLED);
+ scheduler.enableAddedControllerServicesAsync(List.of(addedService));
+ verify(delegate).enableControllerServicesAsync(Collections.emptySet());
+ }
+
@Test
void testExistingEnabledServiceReEnabled() {
final ControllerServiceNode enabledService =
createMockService("enabled-svc", ControllerServiceState.ENABLED);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ComponentScheduler.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ComponentScheduler.java
index 93edfbf390b..f0c374f35bd 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ComponentScheduler.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ComponentScheduler.java
@@ -33,6 +33,24 @@ public interface ComponentScheduler {
void enableControllerServicesAsync(Collection<ControllerServiceNode>
controllerServices);
+ /**
+ * Called during a flow synchronization with the Controller Services that
are newly added by the synchronization
+ * (i.e., that did not exist in the flow before). Because Controller
Services are always persisted in a versioned
+ * flow as DISABLED, they are never included in the "proposed state is
ENABLED" set that is handed to
+ * {@link #enableControllerServicesAsync(Collection)}. This method gives a
scheduler the opportunity to enable
+ * newly added services based on its own policy.
+ *
+ * <p>The default implementation is a no-op: schedulers that restore
components to a persisted state (startup,
+ * cluster reconnect) enable services through their normal proposed-state
handling and must not enable a service
+ * that was not previously enabled. Only a scheduler that retains the
existing runtime state during an update
+ * (a versioned-flow upgrade) overrides this to enable newly added
services when the process group is active,
+ * mirroring how newly added processors are started.</p>
+ *
+ * @param controllerServices the Controller Services that were newly added
by the synchronization
+ */
+ default void
enableAddedControllerServicesAsync(Collection<ControllerServiceNode>
controllerServices) {
+ }
+
void disableControllerServicesAsync(Collection<ControllerServiceNode>
controllerServices);
void startReportingTask(ReportingTaskNode reportingTask);
diff --git
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/CountFlowFiles.java
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/CountFlowFiles.java
index c067ffb622e..c4756e15525 100644
---
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/CountFlowFiles.java
+++
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/CountFlowFiles.java
@@ -41,13 +41,22 @@ public class CountFlowFiles extends AbstractProcessor {
.identifiesControllerService(CountService.class)
.build();
+ static final PropertyDescriptor SECONDARY_COUNT_SERVICE = new Builder()
+ .name("Secondary Count Service")
+ .displayName("Secondary Count Service")
+ .description("An optional additional Controller Service to reference.
This mirrors a new optional controller service "
+ + "reference being introduced on an existing processor in a later
flow version.")
+ .required(false)
+ .identifiesControllerService(CountService.class)
+ .build();
+
public static final Relationship REL_SUCCESS = new Relationship.Builder()
.name("success")
.build();
@Override
protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
- return Collections.singletonList(COUNT_SERVICE);
+ return List.of(COUNT_SERVICE, SECONDARY_COUNT_SERVICE);
}
@Override
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/registry/RegistryClientIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/registry/RegistryClientIT.java
index 4903372c5b2..0d37673225b 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/registry/RegistryClientIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/registry/RegistryClientIT.java
@@ -550,6 +550,105 @@ public class RegistryClientIT extends NiFiSystemIT {
}
}
+ /**
+ * Tests that when upgrading an active versioned process group, a newly
added Controller Service that is referenced only
+ * by an EXISTING processor (via a newly added optional property) is
enabled by the upgrade.
+ *
+ * This is the scenario observed in production with the Openflow
MySQL/PostgreSQL connectors: version N+1 introduced a
+ * new optional Controller Service plus a new optional property on an
already-existing processor that references it.
+ * Controller Services are always persisted in a versioned flow as
DISABLED (see
+ * {@code VersionedComponentStateLookup.ENABLED_OR_DISABLED}), so the
upgrade itself must enable the newly added service
+ * for the referencing processor to remain valid and running.
+ *
+ * It differs from {@link #testNewComponentsStartedDuringVersionChange()}
in one key way: there, the new service is
+ * referenced by a brand-new processor that the upgrade starts. Here, NO
new component references the new service - only
+ * an existing processor does, through a new optional property.
+ *
+ * v1: GenerateFlowFile -> CountFlowFiles (references "Count Service") ->
TerminateFlowFile, plus a standalone Heartbeat.
+ * v2: Adds a new StandardCountService and sets the existing
CountFlowFiles processor's optional "Secondary Count Service"
+ * property to reference it. A standalone Heartbeat GenerateFlowFile
stays running so the group is active during the
+ * upgrade.
+ */
+ @Test
+ public void
testNewServiceReferencedByExistingProcessorEnabledDuringVersionChange() throws
NiFiClientException, IOException, InterruptedException {
+ final FlowRegistryClientEntity clientEntity = registerClient();
+ final NiFiClientUtil util = getClientUtil();
+
+ final ProcessGroupEntity group = util.createProcessGroup("Parent",
"root");
+ final ControllerServiceEntity countService =
util.createControllerService("StandardCountService", group.getId());
+
+ final ProcessorEntity generate =
util.createProcessor("GenerateFlowFile", group.getId());
+ final ProcessorEntity countProcessor =
util.createProcessor("CountFlowFiles", group.getId());
+ util.updateProcessorProperties(countProcessor,
Collections.singletonMap("Count Service", countService.getComponent().getId()));
+
+ final ProcessorEntity terminate =
util.createProcessor("TerminateFlowFile", group.getId());
+ final ConnectionEntity connectionToTerminate =
util.createConnection(countProcessor, terminate, "success");
+ util.setFifoPrioritizer(connectionToTerminate);
+ util.createConnection(generate, countProcessor, "success");
+
+ // Standalone "Heartbeat" processor: present and unchanged in both v1
and v2, stays running during the upgrade so
+ // that the process group is active when the synchronizer inspects it.
+ final ProcessorEntity heartbeat =
util.createProcessor("GenerateFlowFile", group.getId());
+ util.setAutoTerminatedRelationships(heartbeat, "success");
+
+ // Save as v1
+ final VersionControlInformationEntity vci =
util.startVersionControl(group, clientEntity, TEST_FLOWS_BUCKET,
"ExistingProcessorNewService");
+
+ // Build v2: add a new service and reference it from the EXISTING
processor via a new optional property.
+ // No new component references the new service.
+ final ControllerServiceEntity newCountService =
util.createControllerService("StandardCountService", group.getId());
+ util.updateProcessorProperties(countProcessor,
Collections.singletonMap("Secondary Count Service",
newCountService.getComponent().getId()));
+
+ // Save as v2
+ util.saveFlowVersion(group, clientEntity, vci);
+
+ // Switch back to v1 and start the flow
+ util.changeFlowVersion(group.getId(), "1");
+ util.assertFlowStaleAndUnmodified(group.getId());
+
+ util.enableControllerService(countService);
+ util.waitForValidProcessor(generate.getId());
+ util.startProcessor(generate);
+ util.waitForValidProcessor(countProcessor.getId());
+ util.startProcessor(countProcessor);
+ util.waitForValidProcessor(heartbeat.getId());
+ util.startProcessor(heartbeat);
+
+ // Verify v1 flow works
+ waitForQueueCount(connectionToTerminate.getId(), getNumberOfNodes());
+ final Map<String, String> v1Attributes =
util.getQueueFlowFile(connectionToTerminate.getId(),
0).getFlowFile().getAttributes();
+ assertEquals("1", v1Attributes.get("count"));
+
+ // Upgrade to v2 while the flow is running. The Heartbeat processor
stays running, so the process group is active.
+ util.changeFlowVersion(group.getId(), "2");
+ util.assertFlowUpToDate(group.getId());
+
+ // The newly added service is referenced only by the existing
CountFlowFiles processor via the new optional property.
+ // Existing component ids are stable across a version change, so the
newly added service is the one whose id is not
+ // the pre-existing countService.
+ final Set<ControllerServiceEntity> servicesAfterUpgrade =
getNifiClient().getFlowClient()
+ .getControllerServices(group.getId()).getControllerServices();
+ assertEquals(2, servicesAfterUpgrade.size(), "Expected the
pre-existing and the newly added Controller Service");
+ final ControllerServiceEntity newServiceAfterUpgrade =
servicesAfterUpgrade.stream()
+ .filter(service -> !service.getId().equals(countService.getId()))
+ .findFirst()
+ .orElseThrow();
+
+ // Because the group was active during the upgrade, the newly added
service must be enabled so that the existing
+ // processor that references it stays valid and running. Give
enablement a chance to complete, then assert
+ // explicitly: waitForControllerServiceState returns without throwing
on timeout, so an explicit assertion is
+ // needed to surface the failure clearly (rather than only as a
downstream "processor never started" timeout).
+ util.waitForControllerServicesEnabled(group.getId(),
newServiceAfterUpgrade.getId());
+ final ControllerServiceEntity newServiceState =
getNifiClient().getControllerServicesClient()
+ .getControllerService(newServiceAfterUpgrade.getId());
+ assertEquals("ENABLED", newServiceState.getComponent().getState(),
+ "Controller Service '" + newServiceState.getComponent().getName()
+ "' (" + newServiceState.getId() + "), newly added and "
+ + "referenced only by an existing processor via a new optional
property, should be ENABLED after an active-group upgrade");
+
+ // The existing processor that references the newly added service must
be valid and running again.
+ util.waitForRunningProcessor(countProcessor.getId());
+ }
+
/**
* Test that a parent Process Group can be committed to a Flow Registry
even when a child Process Group
* that is separately under Version Control has local modifications. The
parent's snapshot only stores
