Skip to content

Commit 719d1f4

Browse files
adwk67adwk67
committed
add patch for sibling issue NIFI-15801 before (updated) NIFI-15901 patch: version 2.7.2
1 parent d013bb6 commit 719d1f4

3 files changed

Lines changed: 128 additions & 177 deletions

nifi/stackable/patches/2.6.0/0008-NIFI-15901-check-services-are-disabled-before-calling-updateCon.patch renamed to nifi/stackable/patches/2.7.2/0008-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 f19aa11944fa70bf6706aeb619813faece1a606d Mon Sep 17 00:00:00 2001
1+
From 7dd6e0295bce03dcc86a0c671a73b67641c7af25 Mon Sep 17 00:00:00 2001
22
From: Andrew Kenworthy <andrew.kenworthy@stackable.tech>
3-
Date: Mon, 4 May 2026 11:33:36 +0200
4-
Subject: check services are disabled before calling updateControllerService
3+
Date: Tue, 5 May 2026 17:19:10 +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 b79ee4d6e8..0310e57be9 100644
11+
index b79ee4d6e8..d8f2d2da6d 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
@@ -269,8 +269,8 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
@@ -51,71 +51,67 @@ index b79ee4d6e8..0310e57be9 100644
5151

5252
for (final VersionedProcessGroup proposedChildGroup : proposed.getProcessGroups()) {
5353
final ProcessGroup childGroup = childGroupsByVersionedId.get(proposedChildGroup.getIdentifier());
54-
@@ -707,7 +707,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-
@@ -741,17 +741,43 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
64-
updateControllerService(addedService, proposedService, topLevelGroup);
65-
}
54+
@@ -1189,21 +1189,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-
@@ -1375,7 +1401,7 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
107+
}
108+
@@ -1375,7 +1393,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 ProcessGroup group = context.getFlowManager().createProcessGroup(id);
121117
group.setVersionedComponentId(proposed.getIdentifier());

0 commit comments

Comments
 (0)