You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jo...@apache.org on 2016/07/29 15:24:12 UTC
[1/2] nifi git commit: NIFI-2431: Before registering for the Cluster
Coordinator role, check if another node already has this role. If so,
do not register for this role until after the node has joined the cluster and
inherited the flow.
Repository: nifi
Updated Branches:
refs/heads/master cfc8a9613 -> 70d70732b
NIFI-2431: Before registering for the Cluster Coordinator role, check if another node already has this role. If so, do not register for this role until after the node has joined the cluster and inherited the flow.
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/bc0b5fe1
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/bc0b5fe1
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/bc0b5fe1
Branch: refs/heads/master
Commit: bc0b5fe154c8473220c5a119434e79ca279be2c3
Parents: cfc8a96
Author: Mark Payne <ma...@hotmail.com>
Authored: Fri Jul 29 09:14:17 2016 -0400
Committer: joewitt <jo...@apache.org>
Committed: Fri Jul 29 09:42:38 2016 -0400
----------------------------------------------------------------------
.../apache/nifi/controller/FlowController.java | 44 ++++++++++++++------
.../nifi/controller/StandardFlowService.java | 4 +-
.../election/CuratorLeaderElectionManager.java | 3 +-
3 files changed, 36 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/bc0b5fe1/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index 2649895..9ca8e30 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -571,7 +571,23 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
if (configuredForClustering) {
leaderElectionManager = new CuratorLeaderElectionManager(4);
heartbeater = new ClusterProtocolHeartbeater(protocolSender, properties);
- registerForClusterCoordinator();
+
+ // Check if there is already a cluster coordinator elected. If not, go ahead
+ // and register for coordinator role. If there is already one elected, do not register until
+ // we have connected to the cluster. This allows us to avoid becoming the coordinator with a
+ // flow that is different from the rest of the cluster (especially an empty flow) and then
+ // kicking everyone out. This way, we instead inherit the cluster flow before we attempt to be
+ // the coordinator.
+ LOG.info("Checking if there is already a Cluster Coordinator Elected...");
+ final NodeIdentifier electedCoordinatorNodeId = clusterCoordinator.getElectedActiveCoordinatorNode();
+ if (electedCoordinatorNodeId == null) {
+ LOG.info("It appears that no Cluster Coordinator has been Elected yet. Registering for Cluster Coordinator Role.");
+ registerForClusterCoordinator();
+ } else {
+ LOG.info("The Elected Cluster Coordinator is {}. Will not register to be elected for this role until after connecting "
+ + "to the cluster and inheriting the cluster's flow.", electedCoordinatorNodeId);
+ }
+
leaderElectionManager.start();
} else {
leaderElectionManager = null;
@@ -3279,6 +3295,20 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
});
}
+ private void registerForPrimaryNode() {
+ leaderElectionManager.register(ClusterRoles.PRIMARY_NODE, new LeaderElectionStateChangeListener() {
+ @Override
+ public void onLeaderElection() {
+ setPrimary(true);
+ }
+
+ @Override
+ public void onLeaderRelinquish() {
+ setPrimary(false);
+ }
+ });
+ }
+
/**
* Sets whether this instance is clustered. Clustered means that a node is either connected or trying to connect to the cluster.
*
@@ -3315,17 +3345,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
// update the bulletin repository
if (isChanging) {
if (clustered) {
- leaderElectionManager.register(ClusterRoles.PRIMARY_NODE, new LeaderElectionStateChangeListener() {
- @Override
- public void onLeaderElection() {
- setPrimary(true);
- }
-
- @Override
- public void onLeaderRelinquish() {
- setPrimary(false);
- }
- });
+ registerForPrimaryNode();
// Participate in Leader Election for Heartbeat Monitor. Start the heartbeat monitor
// if/when we become leader and stop it when we lose leader role
http://git-wip-us.apache.org/repos/asf/nifi/blob/bc0b5fe1/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
index 71d66b5..1b26e40 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
@@ -772,12 +772,12 @@ public class StandardFlowService implements FlowService, ProtocolHandler {
try {
response = senderListener.requestConnection(requestMsg).getConnectionResponse();
if (response.getRejectionReason() != null) {
- logger.warn("Connection request was blocked by cluster manager with the explanation: " + response.getRejectionReason());
+ logger.warn("Connection request was blocked by cluster coordinator with the explanation: " + response.getRejectionReason());
// set response to null and treat a firewall blockage the same as getting no response from manager
response = null;
break;
} else if (response.shouldTryLater()) {
- logger.info("Flow controller requested by cluster manager to retry connection in " + response.getTryLaterSeconds() + " seconds.");
+ logger.info("Flow controller requested by cluster coordinator to retry connection in " + response.getTryLaterSeconds() + " seconds.");
try {
Thread.sleep(response.getTryLaterSeconds() * 1000);
} catch (final InterruptedException ie) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/bc0b5fe1/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/leader/election/CuratorLeaderElectionManager.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/leader/election/CuratorLeaderElectionManager.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/leader/election/CuratorLeaderElectionManager.java
index bea643f..19fa225 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/leader/election/CuratorLeaderElectionManager.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/leader/election/CuratorLeaderElectionManager.java
@@ -90,7 +90,7 @@ public class CuratorLeaderElectionManager implements LeaderElectionManager {
logger.debug("{} Registering new Leader Selector for role {}", this, roleName);
if (leaderRoles.containsKey(roleName)) {
- logger.warn("{} Attempted to register Leader Election for role '{}' but this role is already registered", this, roleName);
+ logger.info("{} Attempted to register Leader Election for role '{}' but this role is already registered", this, roleName);
return;
}
@@ -130,6 +130,7 @@ public class CuratorLeaderElectionManager implements LeaderElectionManager {
}
leaderSelector.close();
+ logger.info("This node is no longer registered to be elected as the Leader for Role '{}'", roleName);
}
@Override
[2/2] nifi git commit: NIFI-2388: Addressed a spot that assumes that
Process Group is always set on StandardControllerServiceNode
Posted by jo...@apache.org.
NIFI-2388: Addressed a spot that assumes that Process Group is always set on StandardControllerServiceNode
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/70d70732
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/70d70732
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/70d70732
Branch: refs/heads/master
Commit: 70d70732b510283e8b22c603ab7db72589102d1e
Parents: bc0b5fe
Author: Mark Payne <ma...@hotmail.com>
Authored: Fri Jul 29 10:57:33 2016 -0400
Committer: joewitt <jo...@apache.org>
Committed: Fri Jul 29 11:02:52 2016 -0400
----------------------------------------------------------------------
.../service/StandardControllerServiceNode.java | 12 ++++++------
1 file changed, 6 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/70d70732/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
index 7952d0f..0c1e6b3 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
@@ -168,12 +168,12 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i
@Override
public List<ControllerServiceNode> getRequiredControllerServices() {
Set<ControllerServiceNode> requiredServices = new HashSet<>();
- for (Entry<PropertyDescriptor, String> pEntry : this.getProperties().entrySet()) {
- PropertyDescriptor descriptor = pEntry.getKey();
- if (descriptor.getControllerServiceDefinition() != null && pEntry.getValue() != null) {
- ControllerServiceNode rNode = this.processGroup.getControllerService(pEntry.getValue());
- requiredServices.add(rNode);
- requiredServices.addAll(rNode.getRequiredControllerServices());
+ for (Entry<PropertyDescriptor, String> entry : getProperties().entrySet()) {
+ PropertyDescriptor descriptor = entry.getKey();
+ if (descriptor.getControllerServiceDefinition() != null && entry.getValue() != null) {
+ ControllerServiceNode requiredNode = serviceProvider.getControllerServiceNode(entry.getValue());
+ requiredServices.add(requiredNode);
+ requiredServices.addAll(requiredNode.getRequiredControllerServices());
}
}
return new ArrayList<>(requiredServices);