You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by markap14 <gi...@git.apache.org> on 2016/08/15 20:38:24 UTC
[GitHub] nifi pull request #866: NIFI-2566: Refactoring to improve robustness of clus...
GitHub user markap14 opened a pull request:
https://github.com/apache/nifi/pull/866
NIFI-2566: Refactoring to improve robustness of cluster
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/markap14/nifi NIFI-2566
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/nifi/pull/866.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #866
----
commit 90dc74627157296b4084ed65c1b752bfe25db1e8
Author: Mark Payne <ma...@hotmail.com>
Date: 2016-08-13T23:38:07Z
NIFI-2566: Refactored to allow just the Leader Election Manager to be responsible for determining who is the Cluster Coordinator
commit 3fd121b2ff57c5fc51171791351580c29c58e553
Author: Mark Payne <ma...@hotmail.com>
Date: 2016-08-14T23:40:31Z
NIFI-2566: Removed storage of cluster roles from heartbeats and NodeConnectionStatus; use LeaderElectionManager to determine roles instead
commit 457a2df5c3296013ce896ca7900807d9bdb69a71
Author: Mark Payne <ma...@hotmail.com>
Date: 2016-08-15T20:35:56Z
NIFI-2566: Updated Heartbeats so that if a node is out-of-sync with cluster topology, cluster coordinator will provide updated information back to the nodes
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #866: NIFI-2566: Refactoring to improve robustness of cluster
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on the issue:
https://github.com/apache/nifi/pull/866
+1
Visually verified code, and did a contrib check build rebased to master. In a 3 node secure cluster did all sorts of tests involving connecting, disconnecting, restarting, and starting/stopping components. All behaved as expected. Thanks @markap14, I will merge it in.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #866: NIFI-2566: Refactoring to improve robustness of clus...
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/866#discussion_r74833401
--- Diff: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/cluster/ClusterProtocolHeartbeater.java ---
@@ -18,102 +18,80 @@
package org.apache.nifi.controller.cluster;
import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.util.Properties;
-
-import org.apache.curator.RetryPolicy;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.CuratorFrameworkFactory;
-import org.apache.curator.retry.RetryNTimes;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import org.apache.nifi.cluster.coordination.ClusterCoordinator;
+import org.apache.nifi.cluster.coordination.node.ClusterRoles;
+import org.apache.nifi.cluster.coordination.node.NodeConnectionStatus;
+import org.apache.nifi.cluster.protocol.HeartbeatPayload;
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
import org.apache.nifi.cluster.protocol.NodeProtocolSender;
import org.apache.nifi.cluster.protocol.ProtocolException;
import org.apache.nifi.cluster.protocol.message.HeartbeatMessage;
-import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.Watcher;
+import org.apache.nifi.cluster.protocol.message.HeartbeatResponseMessage;
+import org.apache.nifi.controller.leader.election.LeaderElectionManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * Uses ZooKeeper in order to determine which node is the elected Cluster Coordinator and to indicate
- * that this node is part of the cluster. However, once the Cluster Coordinator is known, heartbeats are
+ * Uses Leader Election Manager in order to determine which node is the elected Cluster Coordinator and to indicate
+ * that this node is part of the cluster. Once the Cluster Coordinator is known, heartbeats are
* sent directly to the Cluster Coordinator.
*/
public class ClusterProtocolHeartbeater implements Heartbeater {
private static final Logger logger = LoggerFactory.getLogger(ClusterProtocolHeartbeater.class);
private final NodeProtocolSender protocolSender;
- private final CuratorFramework curatorClient;
- private final String nodesPathPrefix;
-
- private final String coordinatorPath;
- private volatile String coordinatorAddress;
+ private final LeaderElectionManager electionManager;
+ private final ClusterCoordinator clusterCoordinator;
-
- public ClusterProtocolHeartbeater(final NodeProtocolSender protocolSender, final Properties properties) {
+ public ClusterProtocolHeartbeater(final NodeProtocolSender protocolSender, final ClusterCoordinator clusterCoordinator, final LeaderElectionManager electionManager) {
this.protocolSender = protocolSender;
-
- final RetryPolicy retryPolicy = new RetryNTimes(10, 500);
- final ZooKeeperClientConfig zkConfig = ZooKeeperClientConfig.createConfig(properties);
-
- curatorClient = CuratorFrameworkFactory.newClient(zkConfig.getConnectString(),
- zkConfig.getSessionTimeoutMillis(), zkConfig.getConnectionTimeoutMillis(), retryPolicy);
-
- curatorClient.start();
- nodesPathPrefix = zkConfig.resolvePath("cluster/nodes");
- coordinatorPath = nodesPathPrefix + "/coordinator";
+ this.clusterCoordinator = clusterCoordinator;
+ this.electionManager = electionManager;
}
@Override
public String getHeartbeatAddress() throws IOException {
- final String curAddress = coordinatorAddress;
- if (curAddress != null) {
- return curAddress;
+ final String heartbeatAddress = electionManager.getLeader(ClusterRoles.CLUSTER_COORDINATOR);
+ if (heartbeatAddress == null) {
+ throw new ProtocolException("Cannot send heartbeat because there is no Cluster Coordinator currently elected");
}
- try {
- // Get coordinator address and add watcher to change who we are heartbeating to if the value changes.
- final byte[] coordinatorAddressBytes = curatorClient.getData().usingWatcher(new Watcher() {
- @Override
- public void process(final WatchedEvent event) {
- coordinatorAddress = null;
- }
- }).forPath(coordinatorPath);
- final String address = coordinatorAddress = new String(coordinatorAddressBytes, StandardCharsets.UTF_8);
-
- logger.info("Determined that Cluster Coordinator is located at {}; will use this address for sending heartbeat messages", address);
- return address;
- } catch (Exception e) {
- throw new IOException("Unable to determine Cluster Coordinator from ZooKeeper", e);
- }
+ return heartbeatAddress;
}
-
@Override
public synchronized void send(final HeartbeatMessage heartbeatMessage) throws IOException {
final String heartbeatAddress = getHeartbeatAddress();
-
- try {
- protocolSender.heartbeat(heartbeatMessage, heartbeatAddress);
- } catch (final ProtocolException pe) {
- // a ProtocolException is likely the result of not being able to communicate
- // with the coordinator. If we do get an IOException communicating with the coordinator,
- // it will be the cause of the Protocol Exception. In this case, set coordinatorAddress
- // to null so that we double-check next time that the coordinator has not changed.
- if (pe.getCause() instanceof IOException) {
- coordinatorAddress = null;
+ final HeartbeatResponseMessage responseMessage = protocolSender.heartbeat(heartbeatMessage, heartbeatAddress);
+
+ final byte[] payloadBytes = heartbeatMessage.getHeartbeat().getPayload();
+ final HeartbeatPayload payload = HeartbeatPayload.unmarshal(payloadBytes);
+ final List<NodeConnectionStatus> nodeStatusList = payload.getClusterStatus();
+ final Map<NodeIdentifier, Long> updateIdMap = nodeStatusList.stream().collect(
+ Collectors.toMap(status -> status.getNodeIdentifier(), status -> status.getUpdateIdentifier()));
+
+ final List<NodeConnectionStatus> updatedStatuses = responseMessage.getUpdatedNodeStatuses();
+ if (updatedStatuses != null) {
+ for (final NodeConnectionStatus updatedStatus : updatedStatuses) {
+ final NodeIdentifier nodeId = updatedStatus.getNodeIdentifier();
+ final Long updateId = updateIdMap.get(nodeId);
+
+ final boolean updated = clusterCoordinator.resetNodeStatus(updatedStatus, updateId == null ? -1L : updateId);
+ if (updated) {
+ // TODO - REMOVE THE ASTERISKS
--- End diff --
Lingering Todo, can it be removed?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #866: NIFI-2566: Refactoring to improve robustness of cluster
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on the issue:
https://github.com/apache/nifi/pull/866
@markap14, I'm having trouble running a standalone instance with this PR. I get the at I can't start the FlowController due to (very long message, took the end):
`Caused by: java.lang.IllegalStateException: The 'nifi.zookeeper.connect.string' property is not set in nifi.properties
at org.apache.nifi.controller.cluster.ZooKeeperClientConfig.createConfig(ZooKeeperClientConfig.java:76) ~[nifi-framework-core-1.0.0-SNAPSHOT.jar:1.0.0-SNAPSHOT]
at org.apache.nifi.controller.leader.election.CuratorLeaderElectionManager.<init>(CuratorLeaderElectionManager.java:61) ~[nifi-framework-core-1.0.0-SNAPSHOT.jar:1.0.0-SNAPSHOT]
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) ~[na:1.8.0_74]
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62) ~[na:1.8.0_74]
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) ~[na:1.8.0_74]
at java.lang.reflect.Constructor.newInstance(Constructor.java:423) ~[na:1.8.0_74]
at org.springframework.beans.BeanUtils.instantiateClass(BeanUtils.java:147) ~[spring-beans-4.2.4.RELEASE.jar:4.2.4.RELEASE]
... 52 common frames omitted`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #866: NIFI-2566: Refactoring to improve robustness of cluster
Posted by markap14 <gi...@git.apache.org>.
Github user markap14 commented on the issue:
https://github.com/apache/nifi/pull/866
@JPercivall I pushed an update that should address this issue.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #866: NIFI-2566: Refactoring to improve robustness of clus...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/nifi/pull/866
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---