Skip to content

Commit d013bb6

Browse files
adwk67adwk67
committed
add patch for sibling issue NIFI-15801 before (updated) NIFI-15901 patch
1 parent 32b0272 commit d013bb6

2 files changed

Lines changed: 128 additions & 56 deletions

File tree

nifi/stackable/patches/2.9.0/0006-NIFI-15901-check-services-are-disabled-before-calling-updateCont.patch renamed to nifi/stackable/patches/2.9.0/0006-NIFI-15801-Stop-processors-in-synchronizeProcessors-.patch

Lines changed: 52 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,14 @@
1-
From 8f7007ac0ac2f83e27666d865727335156b5053b Mon Sep 17 00:00:00 2001
1+
From 2cf0721126a21eb429500000de96372272d9d3b5 Mon Sep 17 00:00:00 2001
22
From: Andrew Kenworthy <andrew.kenworthy@stackable.tech>
3-
Date: Wed, 29 Apr 2026 17:32:02 +0200
4-
Subject: check services are disabled before calling updateControllerService
3+
Date: Tue, 5 May 2026 16:48:22 +0200
4+
Subject: NIFI-15801 Stop processors in synchronizeProcessors before updating
55

66
---
7-
...tandardVersionedComponentSynchronizer.java | 54 ++++++++++++++-----
8-
1 file changed, 40 insertions(+), 14 deletions(-)
7+
...tandardVersionedComponentSynchronizer.java | 61 ++++++++++++-------
8+
1 file changed, 40 insertions(+), 21 deletions(-)
99

1010
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
11-
index 092d2f7e7b..7cbcc8c6dc 100644
11+
index 092d2f7e7b..2c64fa8cae 100644
1212
--- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
1313
+++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
1414
@@ -270,8 +270,8 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
@@ -51,71 +51,67 @@ index 092d2f7e7b..7cbcc8c6dc 100644
5151

5252
for (final VersionedProcessGroup proposedChildGroup : proposed.getProcessGroups()) {
5353
final ProcessGroup childGroup = childGroupsByVersionedId.get(proposedChildGroup.getIdentifier());
54-
@@ -711,7 +711,7 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
55-
}
56-
57-
private void synchronizeControllerServices(final ProcessGroup group, final VersionedProcessGroup proposed, final Map<String, ControllerServiceNode> servicesByVersionedId,
58-
- final ProcessGroup topLevelGroup) {
59-
+ final ProcessGroup topLevelGroup) throws FlowSynchronizationException {
60-
// Controller Services have to be handled a bit differently than other components. This is because Processors and Controller
61-
// Services may reference other Controller Services. Since we may be adding Service A, which depends on Service B, before adding
62-
// Service B, we need to ensure that we create all Controller Services first and then call updateControllerService for each
63-
@@ -745,17 +745,43 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
64-
updateControllerService(addedService, proposedService, topLevelGroup);
65-
}
54+
@@ -1193,21 +1193,39 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
6655

67-
- // Update all of the Controller Services to match the VersionedControllerService
68-
+ // Update all Controller Services to match the VersionedControllerService.
69-
+ // Services may be ENABLED here if the outer "affected components" pass did not
70-
+ // disable them (e.g. COMPONENT_ADDED diffs are skipped by AffectedComponentSet).
71-
+ // We must disable before calling updateControllerService, which calls setProperties
72-
+ // which calls verifyModifiable and throws IllegalStateException on ENABLED services.
73-
+ final long stopTimeout = System.currentTimeMillis() + syncOptions.getComponentStopTimeout().toMillis();
74-
for (final Map.Entry<ControllerServiceNode, VersionedControllerService> entry : services.entrySet()) {
75-
final ControllerServiceNode service = entry.getKey();
76-
final VersionedControllerService proposedService = entry.getValue();
56+
private void synchronizeProcessors(final ProcessGroup group, final VersionedProcessGroup proposed, final Map<String, ProcessorNode> processorsByVersionedId,
57+
final ProcessGroup topLevelGroup)
58+
- throws ProcessorInstantiationException {
59+
+ throws ProcessorInstantiationException, FlowSynchronizationException {
7760

78-
if (updatedVersionedComponentIds.contains(proposedService.getIdentifier())) {
79-
- updateControllerService(service, proposedService, topLevelGroup);
61+
- for (final VersionedProcessor proposedProcessor : proposed.getProcessors()) {
62+
- final ProcessorNode processor = processorsByVersionedId.get(proposedProcessor.getIdentifier());
63+
- if (processor == null) {
64+
- final ProcessorNode added = addProcessor(group, proposedProcessor, context.getComponentIdGenerator(), topLevelGroup);
65+
- LOG.info("Added {} to {}", added, group);
66+
- } else if (updatedVersionedComponentIds.contains(proposedProcessor.getIdentifier())) {
67+
- updateProcessor(processor, proposedProcessor, topLevelGroup);
8068
- // Any existing component that is modified during synchronization may have its properties reverted to a pre-migration state,
8169
- // so we then add it to the set to allow migrateProperties to be called again to get it back to the migrated state
82-
- createdAndModifiedExtensions.add(new CreatedOrModifiedExtension(service, getPropertyValues(service)));
83-
- LOG.info("Updated {}", service);
84-
+ final Set<ComponentNode> referencesToRestart = new HashSet<>();
85-
+ final Set<ControllerServiceNode> servicesToRestart = new HashSet<>();
70+
- createdAndModifiedExtensions.add(new CreatedOrModifiedExtension(processor, getPropertyValues(processor)));
71+
- LOG.info("Updated {}", processor);
72+
- } else {
73+
- processor.setPosition(new Position(proposedProcessor.getPosition().getX(), proposedProcessor.getPosition().getY()));
74+
+ final Set<ProcessorNode> processorsToRestart = new HashSet<>();
8675
+
87-
+ try {
76+
+ try {
77+
+ for (final VersionedProcessor proposedProcessor : proposed.getProcessors()) {
78+
+ final ProcessorNode processor = processorsByVersionedId.get(proposedProcessor.getIdentifier());
79+
+ if (processor == null) {
80+
+ final ProcessorNode added = addProcessor(group, proposedProcessor, context.getComponentIdGenerator(), topLevelGroup);
81+
+ LOG.info("Added {} to {}", added, group);
82+
+ } else if (updatedVersionedComponentIds.contains(proposedProcessor.getIdentifier())) {
83+
+ final long processorStopDeadline = System.currentTimeMillis() + syncOptions.getComponentStopTimeout().toMillis();
8884
+ try {
89-
+ stopControllerService(service, proposedService, stopTimeout,
90-
+ syncOptions.getComponentStopTimeoutAction(),
91-
+ referencesToRestart, servicesToRestart, syncOptions);
85+
+ final boolean stopped = stopOrTerminate(processor, processorStopDeadline, syncOptions);
86+
+ if (stopped && proposedProcessor.getScheduledState() == org.apache.nifi.flow.ScheduledState.RUNNING) {
87+
+ processorsToRestart.add(processor);
88+
+ }
9289
+ } catch (final TimeoutException e) {
93-
+ throw new FlowSynchronizationException("Failed to stop Controller Service " + service + " in preparation for update", e);
94-
+ } catch (final InterruptedException e) {
95-
+ Thread.currentThread().interrupt();
96-
+ throw new FlowSynchronizationException("Interrupted while stopping Controller Service " + service, e);
97-
+ }
98-
+ updateControllerService(service, proposedService, topLevelGroup);
99-
+ createdAndModifiedExtensions.add(new CreatedOrModifiedExtension(service, getPropertyValues(service)));
100-
+ LOG.info("Updated {}", service);
101-
+ } finally {
102-
+ // Re-enable services and restart components that were stopped for the update,
103-
+ // restoring the controller to its pre-update running state.
104-
+ if (proposedService.getScheduledState() != org.apache.nifi.flow.ScheduledState.DISABLED) {
105-
+ context.getControllerServiceProvider().enableControllerServicesAsync(servicesToRestart);
106-
+ context.getControllerServiceProvider().scheduleReferencingComponents(
107-
+ service, referencesToRestart, context.getComponentScheduler());
90+
+ throw new FlowSynchronizationException("Failed to stop processor " + processor + " in preparation for update", e);
10891
+ }
92+
+ updateProcessor(processor, proposedProcessor, topLevelGroup);
93+
+ // Any existing component that is modified during synchronization may have its properties reverted to a pre-migration state,
94+
+ // so we then add it to the set to allow migrateProperties to be called again to get it back to the migrated state
95+
+ createdAndModifiedExtensions.add(new CreatedOrModifiedExtension(processor, getPropertyValues(processor)));
96+
+ LOG.info("Updated {}", processor);
97+
+ } else {
98+
+ processor.setPosition(new Position(proposedProcessor.getPosition().getX(), proposedProcessor.getPosition().getY()));
10999
+ }
100+
+ }
101+
+ } finally {
102+
+ for (final ProcessorNode processor : processorsToRestart) {
103+
+ processor.getProcessGroup().startProcessor(processor, false);
104+
+ notifyScheduledStateChange((ComponentNode) processor, syncOptions, org.apache.nifi.flow.ScheduledState.RUNNING);
110105
}
111106
}
112-
113-
@@ -1379,7 +1405,7 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
107+
}
108+
@@ -1379,7 +1397,8 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
114109

115110
private ProcessGroup addProcessGroup(final ProcessGroup destination, final VersionedProcessGroup proposed, final ComponentIdGenerator componentIdGenerator,
116111
final Map<String, VersionedParameterContext> versionedParameterContexts,
117112
- final Map<String, ParameterProviderReference> parameterProviderReferences, ProcessGroup topLevelGroup) throws ProcessorInstantiationException {
118-
+ final Map<String, ParameterProviderReference> parameterProviderReferences, ProcessGroup topLevelGroup) throws ProcessorInstantiationException, FlowSynchronizationException {
113+
+ final Map<String, ParameterProviderReference> parameterProviderReferences, ProcessGroup topLevelGroup)
114+
+ throws ProcessorInstantiationException, FlowSynchronizationException {
119115
final String id = componentIdGenerator.generateUuid(proposed.getIdentifier(), proposed.getInstanceIdentifier(), destination.getIdentifier());
120116
final String connectorId = destination.getConnectorIdentifier().orElse(null);
121117
final ProcessGroup group = context.getFlowManager().createProcessGroup(id, connectorId);

‎nifi/stackable/patches/2.9.0/0007-NIFI-15901-Disable-controller-services-before-updati.patch‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
From 18fd66eeba5ebf61048576cc48500611a3e2e5ba Mon Sep 17 00:00:00 2001
2+
From: Andrew Kenworthy <andrew.kenworthy@stackable.tech>
3+
Date: Tue, 5 May 2026 16:57:20 +0200
4+
Subject: NIFI-15901 Disable controller services before updating
5+
6+
---
7+
...tandardVersionedComponentSynchronizer.java | 44 ++++++++++++++++---
8+
1 file changed, 37 insertions(+), 7 deletions(-)
9+
10+
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
11+
index 2c64fa8cae..028dfbf90b 100644
12+
--- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
13+
+++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
14+
@@ -711,7 +711,7 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
15+
}
16+
17+
private void synchronizeControllerServices(final ProcessGroup group, final VersionedProcessGroup proposed, final Map<String, ControllerServiceNode> servicesByVersionedId,
18+
- final ProcessGroup topLevelGroup) {
19+
+ final ProcessGroup topLevelGroup) throws FlowSynchronizationException {
20+
// Controller Services have to be handled a bit differently than other components. This is because Processors and Controller
21+
// Services may reference other Controller Services. Since we may be adding Service A, which depends on Service B, before adding
22+
// Service B, we need to ensure that we create all Controller Services first and then call updateControllerService for each
23+
@@ -745,17 +745,47 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
24+
updateControllerService(addedService, proposedService, topLevelGroup);
25+
}
26+
27+
- // Update all of the Controller Services to match the VersionedControllerService
28+
+ // Update all Controller Services to match the VersionedControllerService.
29+
+ // Services may still be ENABLED here because not all callers disable them before sync-ing.
30+
+ // We must disable before calling updateControllerService, which calls setProperties
31+
+ // which calls verifyModifiable and throws IllegalStateException on ENABLED services.
32+
for (final Map.Entry<ControllerServiceNode, VersionedControllerService> entry : services.entrySet()) {
33+
final ControllerServiceNode service = entry.getKey();
34+
final VersionedControllerService proposedService = entry.getValue();
35+
36+
if (updatedVersionedComponentIds.contains(proposedService.getIdentifier())) {
37+
- updateControllerService(service, proposedService, topLevelGroup);
38+
- // Any existing component that is modified during synchronization may have its properties reverted to a pre-migration state,
39+
- // so we then add it to the set to allow migrateProperties to be called again to get it back to the migrated state
40+
- createdAndModifiedExtensions.add(new CreatedOrModifiedExtension(service, getPropertyValues(service)));
41+
- LOG.info("Updated {}", service);
42+
+ final long stopTimeout = System.currentTimeMillis() + syncOptions.getComponentStopTimeout().toMillis();
43+
+ final Set<ComponentNode> referencesToRestart = new HashSet<>();
44+
+ final Set<ControllerServiceNode> servicesToRestart = new HashSet<>();
45+
+
46+
+ try {
47+
+ try {
48+
+ stopControllerService(service, proposedService, stopTimeout,
49+
+ syncOptions.getComponentStopTimeoutAction(),
50+
+ referencesToRestart, servicesToRestart, syncOptions);
51+
+ } catch (final TimeoutException e) {
52+
+ throw new FlowSynchronizationException("Failed to stop Controller Service " + service + " in preparation for update", e);
53+
+ } catch (final InterruptedException e) {
54+
+ Thread.currentThread().interrupt();
55+
+ throw new FlowSynchronizationException("Interrupted while stopping Controller Service " + service, e);
56+
+ }
57+
+ updateControllerService(service, proposedService, topLevelGroup);
58+
+ createdAndModifiedExtensions.add(new CreatedOrModifiedExtension(service, getPropertyValues(service)));
59+
+ LOG.info("Updated {}", service);
60+
+ } finally {
61+
+ // Re-enable services and restart components that were stopped for the update,
62+
+ // restoring the controller to its pre-update running state.
63+
+ if (proposedService.getScheduledState() != org.apache.nifi.flow.ScheduledState.DISABLED) {
64+
+ // Use the component scheduler (not the provider directly) which has
65+
+ // already been paused, avoiding a race with the enable loop below.
66+
+ context.getComponentScheduler().enableControllerServicesAsync(servicesToRestart);
67+
+ notifyScheduledStateChange(servicesToRestart, syncOptions, org.apache.nifi.flow.ScheduledState.ENABLED);
68+
+ context.getControllerServiceProvider().scheduleReferencingComponents(
69+
+ service, referencesToRestart, context.getComponentScheduler());
70+
+ referencesToRestart.forEach(componentNode ->
71+
+ notifyScheduledStateChange(componentNode, syncOptions, org.apache.nifi.flow.ScheduledState.RUNNING));
72+
+ }
73+
+ }
74+
}
75+
}
76+

0 commit comments

Comments
 (0)