You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by bb...@apache.org on 2018/03/26 17:08:17 UTC
nifi git commit: NIFI-5012: When connecting to cluster,
esure that controller services appropriately enabled/disabled
Repository: nifi
Updated Branches:
refs/heads/master 3612fbe52 -> 456c9c8fc
NIFI-5012: When connecting to cluster, esure that controller services appropriately enabled/disabled
This closes #2579.
Signed-off-by: Bryan Bende <bb...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/456c9c8f
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/456c9c8f
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/456c9c8f
Branch: refs/heads/master
Commit: 456c9c8fc0f8ebd0131ea02aaa89fc374e3c6882
Parents: 3612fbe
Author: Mark Payne <ma...@hotmail.com>
Authored: Fri Mar 23 15:44:06 2018 -0400
Committer: Bryan Bende <bb...@apache.org>
Committed: Mon Mar 26 13:07:58 2018 -0400
----------------------------------------------------------------------
.../controller/StandardFlowSynchronizer.java | 44 ++++++++++++++++++++
1 file changed, 44 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/456c9c8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
index 425110c..8a79acf 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
@@ -73,6 +73,7 @@ import org.apache.nifi.controller.serialization.FlowSynchronizer;
import org.apache.nifi.controller.serialization.StandardFlowSerializer;
import org.apache.nifi.controller.service.ControllerServiceLoader;
import org.apache.nifi.controller.service.ControllerServiceNode;
+import org.apache.nifi.controller.service.ControllerServiceState;
import org.apache.nifi.encrypt.StringEncryptor;
import org.apache.nifi.events.BulletinFactory;
import org.apache.nifi.fingerprint.FingerprintException;
@@ -749,6 +750,19 @@ public class StandardFlowSynchronizer implements FlowSynchronizer {
}
}
+ private ControllerServiceState getFinalTransitionState(final ControllerServiceState state) {
+ switch (state) {
+ case DISABLED:
+ case DISABLING:
+ return ControllerServiceState.DISABLED;
+ case ENABLED:
+ case ENABLING:
+ return ControllerServiceState.ENABLED;
+ default:
+ throw new AssertionError();
+ }
+ }
+
private ProcessGroup updateProcessGroup(final FlowController controller, final ProcessGroup parentGroup, final Element processGroupElement,
final StringEncryptor encryptor, final FlowEncodingVersion encodingVersion) throws ProcessorInstantiationException {
@@ -779,6 +793,36 @@ public class StandardFlowSynchronizer implements FlowSynchronizer {
// get the real process group and ID
final ProcessGroup processGroup = controller.getGroup(processGroupDto.getId());
+ // determine the scheduled state of all of the Controller Service
+ final List<Element> controllerServiceNodeList = getChildrenByTagName(processGroupElement, "controllerService");
+ final Set<ControllerServiceNode> toDisable = new HashSet<>();
+ final Set<ControllerServiceNode> toEnable = new HashSet<>();
+
+ for (final Element serviceElement : controllerServiceNodeList) {
+ final ControllerServiceDTO dto = FlowFromDOMFactory.getControllerService(serviceElement, encryptor);
+ final ControllerServiceNode serviceNode = processGroup.getControllerService(dto.getId());
+
+ // Check if the controller service is in the correct state. We consider it the correct state if
+ // we are in a transitional state and heading in the right direction or already in the correct state.
+ // E.g., it is the correct state if it should be 'DISABLED' and it is either DISABLED or DISABLING.
+ final ControllerServiceState serviceState = getFinalTransitionState(serviceNode.getState());
+ final ControllerServiceState clusterState = getFinalTransitionState(ControllerServiceState.valueOf(dto.getState()));
+
+ if (serviceState != clusterState) {
+ switch (clusterState) {
+ case DISABLED:
+ toDisable.add(serviceNode);
+ break;
+ case ENABLED:
+ toEnable.add(serviceNode);
+ break;
+ }
+ }
+ }
+
+ controller.disableControllerServicesAsync(toDisable);
+ controller.enableControllerServices(toEnable);
+
// processors & ports cannot be updated - they must be the same. Except for the scheduled state.
final List<Element> processorNodeList = getChildrenByTagName(processGroupElement, "processor");
for (final Element processorElement : processorNodeList) {