ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [38/50] [abbrv] ignite git commit: zk
Date Wed, 20 Dec 2017 08:04:46 GMT
zk


Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/9e4f2c20
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/9e4f2c20
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/9e4f2c20

Branch: refs/heads/ignite-zk
Commit: 9e4f2c20c2582655637c5abf349abc9e5473e9fe
Parents: fc085a4
Author: sboikov <sboikov@gridgain.com>
Authored: Mon Dec 18 15:25:34 2017 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Mon Dec 18 18:08:47 2017 +0300

----------------------------------------------------------------------
 .../internal/ZkCommunicationErrorNodeState.java |   8 +-
 .../ZkCommunicationErrorProcessFuture.java      |  28 ++
 ...kCommunicationErrorResolveFinishMessage.java |   7 +-
 .../ZkCommunicationErrorResolveResult.java      |  15 +-
 .../internal/ZkCommunicationProblemContext.java |  67 +++-
 .../zk/internal/ZkDiscoveryEventsData.java      |   2 +-
 .../ZkDistributedCollectDataFuture.java         |  11 +-
 .../discovery/zk/internal/ZookeeperClient.java  |   2 +-
 .../zk/internal/ZookeeperDiscoveryImpl.java     | 359 +++++++++++++++----
 .../ZookeeperDiscoverySpiBasicTest.java         | 138 ++++++-
 10 files changed, 540 insertions(+), 97 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/9e4f2c20/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorNodeState.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorNodeState.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorNodeState.java
index ddc310d..9c21f13 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorNodeState.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorNodeState.java
@@ -23,21 +23,23 @@ import java.util.BitSet;
 /**
  *
  */
-public class ZkCommunicationErrorNodeState implements Serializable {
+class ZkCommunicationErrorNodeState implements Serializable {
     /** */
     private static final long serialVersionUID = 0L;
 
     /** */
-    private final BitSet commState;
+    final BitSet commState;
 
     /** */
-    private final Exception err;
+    final Exception err;
 
     /**
      * @param commState Communication state.
      * @param err Error if failed get communication state..
      */
     ZkCommunicationErrorNodeState(BitSet commState, Exception err) {
+        assert commState != null || err != null;
+
         this.commState = commState;
         this.err = err;
     }

http://git-wip-us.apache.org/repos/asf/ignite/blob/9e4f2c20/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorProcessFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorProcessFuture.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorProcessFuture.java
index a6294bd..0074817 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorProcessFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorProcessFuture.java
@@ -214,6 +214,34 @@ class ZkCommunicationErrorProcessFuture extends GridFutureAdapter<Void>
implemen
     }
 
     /**
+     * @param err Error.
+     */
+    void onError(Exception err) {
+        assert err != null;
+
+        Map<Long, GridFutureAdapter<Boolean>> futs;
+
+        synchronized (this) {
+            if (state == State.DONE) {
+                assert resErr != null;
+
+                return;
+            }
+
+            state = State.DONE;
+
+            resErr = err;
+
+            futs = nodeFuts; // nodeFuts should not be modified after state changed to DONE.
+        }
+
+        for (Map.Entry<Long, GridFutureAdapter<Boolean>> e : futs.entrySet())
+            e.getValue().onDone(err);
+
+        onDone(err);
+    }
+
+    /**
      * @param failedNodes Node failed as result of resolve process.
      */
     void onFinishResolve(Set<Long> failedNodes) {

http://git-wip-us.apache.org/repos/asf/ignite/blob/9e4f2c20/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveFinishMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveFinishMessage.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveFinishMessage.java
index 20aeddf..147b78f 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveFinishMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveFinishMessage.java
@@ -33,13 +33,18 @@ class ZkCommunicationErrorResolveFinishMessage implements DiscoverySpiCustomMess
     final UUID futId;
 
     /** */
+    final long topVer;
+
+    /** */
     transient ZkCommunicationErrorResolveResult res;
 
     /**
      * @param futId Future ID.
+     * @param topVer Topology version when resolve process finished.
      */
-    ZkCommunicationErrorResolveFinishMessage(UUID futId) {
+    ZkCommunicationErrorResolveFinishMessage(UUID futId, long topVer) {
         this.futId = futId;
+        this.topVer = topVer;
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/9e4f2c20/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveResult.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveResult.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveResult.java
index 607f93b..68cbdb8 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveResult.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveResult.java
@@ -18,7 +18,7 @@
 package org.apache.ignite.spi.discovery.zk.internal;
 
 import java.io.Serializable;
-import org.apache.ignite.internal.util.GridLongList;
+import org.apache.ignite.internal.util.GridIntList;
 import org.jetbrains.annotations.Nullable;
 
 /**
@@ -29,12 +29,17 @@ class ZkCommunicationErrorResolveResult implements Serializable {
     private static final long serialVersionUID = 0L;
 
     /** */
-    final GridLongList failedNodes;
+    final GridIntList killedNodes;
+
+    /** */
+    final Exception err;
 
     /**
-     * @param failedNodes
+     * @param killedNodes Killed nodes.
+     * @param err Error.
      */
-    ZkCommunicationErrorResolveResult(@Nullable GridLongList failedNodes) {
-        this.failedNodes = failedNodes;
+    ZkCommunicationErrorResolveResult(@Nullable GridIntList killedNodes, Exception err) {
+        this.killedNodes = killedNodes;
+        this.err = err;
     }
 }

http://git-wip-us.apache.org/repos/asf/ignite/blob/9e4f2c20/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationProblemContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationProblemContext.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationProblemContext.java
index fd11b55..9cb48cc 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationProblemContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationProblemContext.java
@@ -17,9 +17,14 @@
 
 package org.apache.ignite.spi.discovery.zk.internal;
 
+import java.util.BitSet;
+import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
+import java.util.UUID;
 import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.configuration.CommunicationProblemContext;
 
@@ -28,31 +33,73 @@ import org.apache.ignite.configuration.CommunicationProblemContext;
  */
 class ZkCommunicationProblemContext implements CommunicationProblemContext {
     /** */
+    private static final Comparator<ClusterNode> NODE_ORDER_CMP = new Comparator<ClusterNode>()
{
+        @Override public int compare(ClusterNode node1, ClusterNode node2) {
+            return Long.compare(node1.order(), node2.order());
+        }
+    };
+
+    /** */
     private Set<ClusterNode> killedNodes = new HashSet<>();
 
+    /** */
+    private final Map<UUID, BitSet> nodesState;
+
+    /** */
+    private final List<ClusterNode> initialNodes;
+
+    /** */
+    private final List<ClusterNode> curNodes;
+
+    /**
+     * @param curNodes Current topology snapshot.
+     * @param initialNodes Topology snapshot when communication error resolve started.
+     * @param nodesState Nodes communication state.
+     */
+    ZkCommunicationProblemContext(List<ClusterNode> curNodes,
+        List<ClusterNode> initialNodes,
+        Map<UUID, BitSet> nodesState)
+    {
+        this.curNodes = Collections.unmodifiableList(curNodes);
+        this.initialNodes = initialNodes;
+        this.nodesState = nodesState;
+    }
+
     /** {@inheritDoc} */
     @Override public List<ClusterNode> topologySnapshot() {
-        return null;
+        return curNodes;
     }
 
     /** {@inheritDoc} */
     @Override public boolean connectionAvailable(ClusterNode node1, ClusterNode node2) {
-        return false;
+        BitSet nodeState = nodesState.get(node1.id());
+
+        if (nodeState == null)
+            throw new IllegalArgumentException("Invalid node: " + node1);
+
+        int nodeIdx = Collections.binarySearch(initialNodes, node2, NODE_ORDER_CMP);
+
+        if (nodeIdx < 0)
+            throw new IllegalArgumentException("Invalid node: " + node2);
+
+        assert nodeIdx < nodeState.size() : nodeIdx;
+
+        return nodeState.get(nodeIdx);
     }
 
     /** {@inheritDoc} */
     @Override public List<String> startedCaches() {
-        return null;
+        return null; // TODO ZK
     }
 
     /** {@inheritDoc} */
     @Override public List<List<ClusterNode>> cacheAffinity(String cacheName)
{
-        return null;
+        return null; // TODO ZK
     }
 
     /** {@inheritDoc} */
     @Override public List<List<ClusterNode>> cachePartitionOwners(String cacheName)
{
-        return null;
+        return null; // TODO ZK
     }
 
     /** {@inheritDoc} */
@@ -60,6 +107,16 @@ class ZkCommunicationProblemContext implements CommunicationProblemContext
{
         if (node == null)
             throw new NullPointerException();
 
+        if (Collections.binarySearch(curNodes, node, NODE_ORDER_CMP) < 0)
+            throw new IllegalArgumentException("Invalid node: " + node);
+
         killedNodes.add(node);
     }
+
+    /**
+     * @return Nodes to fail.
+     */
+    Set<ClusterNode> killedNodes() {
+        return killedNodes;
+    }
 }

http://git-wip-us.apache.org/repos/asf/ignite/blob/9e4f2c20/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkDiscoveryEventsData.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkDiscoveryEventsData.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkDiscoveryEventsData.java
index faea49e..6bdf573 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkDiscoveryEventsData.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkDiscoveryEventsData.java
@@ -69,7 +69,7 @@ class ZkDiscoveryEventsData implements Serializable {
     /**
      * @param id Future ID.
      */
-     void communicationErrorResolveFutureId(UUID id) {
+     void communicationErrorResolveFutureId(@Nullable UUID id) {
         commErrFutId = id;
     }
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/9e4f2c20/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkDistributedCollectDataFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkDistributedCollectDataFuture.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkDistributedCollectDataFuture.java
index 19e2acc..2467928 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkDistributedCollectDataFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkDistributedCollectDataFuture.java
@@ -141,9 +141,14 @@ class ZkDistributedCollectDataFuture extends GridFutureAdapter<Void>
{
         // TODO ZK: use multi, better batching + max-size safe + NoNodeException safe.
         String evtDir = paths.distributedFutureBasePath(futId);
 
-        client.deleteAll(evtDir,
-            client.getChildren(evtDir),
-            -1);
+        try {
+            client.deleteAll(evtDir,
+                client.getChildren(evtDir),
+                -1);
+        }
+        catch (KeeperException.NoNodeException e) {
+            // TODO ZK
+        }
 
         client.deleteIfExists(evtDir, -1);
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/9e4f2c20/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperClient.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperClient.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperClient.java
index 5923b39..5c0bd58 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperClient.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperClient.java
@@ -940,7 +940,7 @@ public class ZookeeperClient implements Watcher {
             else if (rc == KeeperException.Code.SESSIONEXPIRED.intValue())
                 U.warn(log, "Failed to execute async operation, connection lost [path=" +
path + ']');
             else
-                assert rc == 0 : rc;
+                assert rc == 0 : KeeperException.Code.get(rc);
         }
     }
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/9e4f2c20/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java
index 62fc581..ef3504f 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java
@@ -19,6 +19,7 @@ package org.apache.ignite.spi.discovery.zk.internal;
 
 import java.io.Serializable;
 import java.util.ArrayList;
+import java.util.BitSet;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
@@ -50,6 +51,7 @@ import org.apache.ignite.internal.IgniteNodeAttributes;
 import org.apache.ignite.internal.IgnitionEx;
 import org.apache.ignite.internal.events.DiscoveryCustomEvent;
 import org.apache.ignite.internal.managers.discovery.IgniteDiscoverySpiInternalListener;
+import org.apache.ignite.internal.util.GridIntList;
 import org.apache.ignite.internal.util.GridSpinBusyLock;
 import org.apache.ignite.internal.util.IgniteUtils;
 import org.apache.ignite.internal.util.future.GridFutureAdapter;
@@ -246,6 +248,8 @@ public class ZookeeperDiscoveryImpl {
      * @param err Connect error.
      */
     public void onCommunicationConnectionError(ClusterNode node0, Exception err) {
+        checkState();
+
         ZookeeperClusterNode node = node(node0.id());
 
         if (node == null)
@@ -269,7 +273,16 @@ public class ZookeeperDiscoveryImpl {
                             ", err= " + err + ']');
                     }
 
-                    fut.scheduleCheckOnTimeout();
+                    ConnectionState connState;
+
+                    synchronized (this) {
+                        connState = this.connState;
+                    }
+
+                    if (connState != ConnectionState.STARTED)
+                        fut.onError(new IgniteCheckedException("Node stopped."));
+                    else
+                        fut.scheduleCheckOnTimeout();
                 }
                 else
                     fut = commErrProcFut.get();
@@ -475,7 +488,7 @@ public class ZookeeperDiscoveryImpl {
                 break;
 
             case STOPPED:
-                throw new IgniteSpiException("Zookeeper client closed.");
+                throw new IgniteSpiException("Node stopped.");
 
             case DISCONNECTED:
                 throw new IgniteClientDisconnectedException(null, "Client is disconnected.");
@@ -1025,11 +1038,41 @@ public class ZookeeperDiscoveryImpl {
     }
 
     /**
+     * @param lastEvts Last events from previous coordinator.
+     * @throws Exception If failed.
+     */
+    private void previousCoordinatorCleanup(ZkDiscoveryEventsData lastEvts) throws Exception
{
+        for (ZkDiscoveryEventData evtData : lastEvts.evts.values()) {
+            if (evtData instanceof ZkDiscoveryCustomEventData) {
+                ZkDiscoveryCustomEventData evtData0 = (ZkDiscoveryCustomEventData)evtData;
+
+                // It is possible previous coordinator failed before finished message processing.
+                if (evtData0.msg instanceof ZkCommunicationErrorResolveFinishMessage) {
+                    try {
+                        ZkCommunicationErrorResolveFinishMessage msg =
+                            (ZkCommunicationErrorResolveFinishMessage)evtData0.msg;
+
+                        ZkCommunicationErrorResolveResult res = unmarshalZip(
+                            ZkDistributedCollectDataFuture.readResult(rtState.zkClient, zkPaths,
msg.futId));
+
+                        deleteAliveNodes(res.killedNodes);
+                    }
+                    catch (KeeperException.NoNodeException ignore) {
+                        // No-op.
+                    }
+                }
+            }
+        }
+    }
+
+    /**
      * @param aliveNodes Alive nodes paths.
      * @param locInternalId Local node's internal ID.
      * @throws Exception If failed.
      */
     private void onBecomeCoordinator(List<String> aliveNodes, int locInternalId) throws
Exception {
+        long topVer0 = rtState.evtsData != null ? rtState.evtsData.topVer : -1L;
+
         byte[] evtsDataBytes = rtState.zkClient.getData(zkPaths.evtsPath);
 
         if (evtsDataBytes.length > 0)
@@ -1044,6 +1087,11 @@ public class ZookeeperDiscoveryImpl {
             assert locNode.order() > 0 : locNode;
             assert rtState.evtsData != null;
 
+            previousCoordinatorCleanup(rtState.evtsData);
+
+            if (topVer0 > rtState.evtsData.topVer)
+                rtState.evtsData.topVer = topVer0;
+
             UUID futId = rtState.evtsData.communicationErrorResolveFutureId();
 
             if (futId != null) {
@@ -1636,6 +1684,7 @@ public class ZookeeperDiscoveryImpl {
 
                                 assert node != null :  msg0.nodeId;
 
+                                // TODO ZK: delete when process event
                                 for (String child : zkClient.getChildren(zkPaths.aliveNodesDir))
{
                                     if (ZkIgnitePaths.aliveInternalId(child) == node.internalId())
{
                                         zkClient.deleteIfExistsAsync(zkPaths.aliveNodesDir
+ "/" + child);
@@ -2021,22 +2070,25 @@ public class ZookeeperDiscoveryImpl {
         ClusterNode creatorNode = rtState.top.nodesById.get(evtData.sndNodeId);
 
         if (msg.warning != null) {
-            U.warn(log, "Received EVT_NODE_FAILED event with warning [" +
-                "nodeInitiatedEvt=" + (creatorNode != null ? creatorNode : evtData.sndNodeId)
+
-                ", nodeId=" + msg.nodeId +
-                ", msg=" + msg.warning + ']');
+            U.warn(log, "Received force EVT_NODE_FAILED event with warning [" +
+                "nodeId=" + msg.nodeId +
+                ", msg=" + msg.warning +
+                ", nodeInitiatedEvt=" + (creatorNode != null ? creatorNode : evtData.sndNodeId)
+ ']');
         }
         else {
             U.warn(log, "Received force EVT_NODE_FAILED event [" +
-                "nodeInitiatedEvt=" + (creatorNode != null ? creatorNode : evtData.sndNodeId)
+
-                ", nodeId=" + msg.nodeId + ']');
+                "nodeId=" + msg.nodeId +
+                ", nodeInitiatedEvt=" + (creatorNode != null ? creatorNode : evtData.sndNodeId)
+ ']');
         }
 
         ZookeeperClusterNode node = rtState.top.nodesById.get(msg.nodeId);
 
         assert node != null : msg.nodeId;
 
-        processNodeFail(node.internalId(), evtData.topologyVersion());
+        if (node.isLocal())
+            throw localNodeFail("Received force EVT_NODE_FAILED event for local node.");
+        else
+            notifyNodeFail(node.internalId(), evtData.topologyVersion());
     }
 
     /**
@@ -2049,13 +2101,10 @@ public class ZookeeperDiscoveryImpl {
         UUID futId = msg.futId;
 
         assert futId != null;
-        assert futId.equals(rtState.evtsData.communicationErrorResolveFutureId());
 
         if (log.isInfoEnabled())
             log.info("Received communication error resolve finish message [reqId=" + futId
+ ']');
 
-        rtState.evtsData.communicationErrorResolveFutureId(null);
-
         rtState.commErrProcNodes = null;
 
         ZkCommunicationErrorResolveResult res = msg.res;
@@ -2069,14 +2118,72 @@ public class ZookeeperDiscoveryImpl {
 
         Set<Long> failedNodes = null;
 
-        if (res.failedNodes != null) {
-            failedNodes = U.newHashSet(res.failedNodes.size());
+        if (res.err != null)
+            U.error(log, "Communication error resolve failed: " + res.err, res.err);
+        else {
+            if (res.killedNodes != null) {
+                failedNodes = U.newHashSet(res.killedNodes.size());
+
+                for (int i = 0; i < res.killedNodes.size(); i++) {
+                    int internalId = res.killedNodes.get(i);
+
+                    if (internalId == locNode.internalId()) {
+                        fut.onError(new IgniteCheckedException("Local node is forced to stop
" +
+                            "by communication error resolver"));
+
+                        if (rtState.crd)
+                            deleteAliveNodes(res.killedNodes);
+
+                        throw localNodeFail("Local node is forced to stop by communication
error resolver " +
+                            "[nodeId=" + locNode.id() + ']');
+                    }
+
+                    ZookeeperClusterNode node = rtState.top.nodesByInternalId.get(internalId);
+
+                    assert node != null : internalId;
+
+                    failedNodes.add(node.order());
+                }
+
+                long topVer = msg.topVer;
+
+                for (int i = 0; i < res.killedNodes.size(); i++) {
+                    int nodeInternalId = res.killedNodes.get(i);
 
-            for (int i = 0; i < res.failedNodes.size(); i++)
-                failedNodes.add(res.failedNodes.get(i));
+                    ClusterNode node = rtState.top.nodesByInternalId.get(nodeInternalId);
+
+                    assert node != null : nodeInternalId;
+
+                    if (log.isInfoEnabled())
+                        log.info("Node stop is forced by communication error resolver [nodeId="
+ node.id() + ']');
+
+                    notifyNodeFail(nodeInternalId, ++topVer);
+                }
+            }
         }
 
         fut.onFinishResolve(failedNodes);
+
+        if (rtState.crd)
+            deleteAliveNodes(res.killedNodes);
+    }
+
+    /**
+     * @param internalIds Nodes internal IDs.
+     * @throws Exception If failed.
+     */
+    private void deleteAliveNodes(@Nullable GridIntList internalIds) throws Exception {
+        if (internalIds == null)
+            return;
+
+        List<String> alives = rtState.zkClient.getChildren(zkPaths.aliveNodesDir);
+
+        for (int i = 0; i < alives.size(); i++) {
+            String alive = alives.get(i);
+
+            if (internalIds.contains(ZkIgnitePaths.aliveInternalId(alive)))
+                rtState.zkClient.deleteIfExistsAsync(zkPaths.aliveNodesDir + "/" + alive);
+        }
     }
 
     /**
@@ -2163,7 +2270,7 @@ public class ZookeeperDiscoveryImpl {
     private void onCommunicationErrorResolveStatusReceived(ZkRuntimeState rtState) throws
Exception {
         ZkDiscoveryEventsData evtsData = rtState.evtsData;
 
-        UUID futId = rtState.evtsData.communicationErrorResolveFutureId();
+        UUID futId = evtsData.communicationErrorResolveFutureId();
 
         if (log.isInfoEnabled())
             log.info("Received communication status from all nodes [reqId=" + futId + ']');
@@ -2178,43 +2285,113 @@ public class ZookeeperDiscoveryImpl {
 
         rtState.commErrProcNodes = null;
 
-        ZkClusterNodes top = rtState.top;
+        List<ClusterNode> topSnapshot = rtState.top.topologySnapshot();
 
-        List<ZkCommunicationErrorNodeState> nodesRes = new ArrayList<>();
+        Map<UUID, BitSet> nodesRes = U.newHashMap(topSnapshot.size());
 
-        for (ZookeeperClusterNode node : top.nodesByOrder.values()) {
+        Exception err = null;
+
+        for (ClusterNode node : topSnapshot) {
             byte[] stateBytes = ZkDistributedCollectDataFuture.readNodeResult(futPath,
                 rtState.zkClient,
                 node.order());
 
             ZkCommunicationErrorNodeState nodeState = unmarshalZip(stateBytes);
 
-            nodesRes.add(nodeState);
+            if (nodeState.err != null) {
+                if (err == null)
+                    err = new Exception("Failed to resolve communication error.");
+
+                err.addSuppressed(nodeState.err);
+            }
+            else {
+                assert nodeState.commState != null;
+
+                nodesRes.put(node.id(), nodeState.commState);
+            }
         }
 
-        ZkCommunicationErrorResolveFinishMessage msg = new ZkCommunicationErrorResolveFinishMessage(futId);
+        long topVer = evtsData.topVer;
 
-        ZkCommunicationErrorResolveResult res = new ZkCommunicationErrorResolveResult(null);
+        GridIntList killedNodesList = null;
 
-        msg.res = res;
+        if (err == null) {
+            boolean fullyConnected = true;
 
-        ZkDistributedCollectDataFuture.saveResult(zkPaths.distributedFutureResultPath(futId),
-            rtState.zkClient,
-            marshalZip(res));
+            for (Map.Entry<UUID, BitSet> e : nodesRes.entrySet()) {
+                if (!checkFullyConnected(e.getValue(), initialNodes, rtState.top)) {
+                    fullyConnected = false;
+
+                    break;
+                }
+            }
 
-        CommunicationProblemResolver rslvr = spi.ignite().configuration().getCommunicationProblemResolver();
+            if (fullyConnected) {
+                if (log.isInfoEnabled()) {
+                    log.info("Finish communication error resolve process automatically, there
are no " +
+                        "communication errors [reqId=" + futId + ']');
+                }
+            }
+            else {
+                CommunicationProblemResolver rslvr = spi.ignite().configuration().getCommunicationProblemResolver();
+
+                if (rslvr != null) {
+                    if (log.isInfoEnabled()) {
+                        log.info("Call communication error resolver [reqId=" + futId +
+                            ", rslvr=" + rslvr.getClass().getSimpleName() + ']');
+                    }
 
-        if (rslvr != null) {
-            ZkCommunicationProblemContext ctx = new ZkCommunicationProblemContext();
+                    ZkCommunicationProblemContext ctx = new ZkCommunicationProblemContext(topSnapshot,
+                        initialNodes,
+                        nodesRes);
 
-            rslvr.resolve(ctx);
+                    try {
+                        rslvr.resolve(ctx);
+
+                        Set<ClusterNode> killedNodes = ctx.killedNodes();
+
+                        if (killedNodes != null) {
+                            if (log.isInfoEnabled()) {
+                                log.info("Communication error resolver forces nodes stop
[reqId=" + futId +
+                                    ", killNodeCnt=" + killedNodes.size() +
+                                    ", nodeIds=" + U.nodeIds(killedNodes) + ']');
+                            }
+
+                            killedNodesList = new GridIntList(killedNodes.size());
+
+                            for (ClusterNode killedNode : killedNodes) {
+                                killedNodesList.add(((ZookeeperClusterNode)killedNode).internalId());
+
+                                evtsData.topVer++;
+                            }
+                        }
+                    }
+                    catch (Exception e) {
+                        err = e;
+
+                        U.error(log, "Failed to resolve communication error with configured
resolver [reqId=" + futId + ']', e);
+                    }
+                }
+            }
         }
 
+        evtsData.communicationErrorResolveFutureId(null);
+
+        ZkCommunicationErrorResolveResult res = new ZkCommunicationErrorResolveResult(killedNodesList,
err);
+
+        ZkCommunicationErrorResolveFinishMessage msg = new ZkCommunicationErrorResolveFinishMessage(futId,
topVer);
+
+        msg.res = res;
+
+        ZkDistributedCollectDataFuture.saveResult(zkPaths.distributedFutureResultPath(futId),
+            rtState.zkClient,
+            marshalZip(res));
+
         evtsData.evtIdGen++;
 
         ZkDiscoveryCustomEventData evtData = new ZkDiscoveryCustomEventData(
             evtsData.evtIdGen,
-            evtsData.topVer,
+            topVer,
             locNode.id(),
             msg,
             null,
@@ -2228,6 +2405,30 @@ public class ZookeeperDiscoveryImpl {
     }
 
     /**
+     * @param commState Node communication state.
+     * @param initialNodes Topology snapshot when communication error resolve started.
+     * @param top Current topology.
+     * @return {@code True} if node has connection to all alive nodes.
+     */
+    private boolean checkFullyConnected(BitSet commState, List<ClusterNode> initialNodes,
ZkClusterNodes top) {
+        int startIdx = 0;
+
+        for (;;) {
+            int idx = commState.nextClearBit(startIdx);
+
+            if (idx >= initialNodes.size())
+                return true;
+
+            ClusterNode node = initialNodes.get(idx);
+
+            if (top.nodesById.containsKey(node.id()))
+                return false;
+
+            startIdx = idx + 1;
+        }
+    }
+
+    /**
      *
      */
     public void simulateNodeFailure() {
@@ -2292,70 +2493,73 @@ public class ZookeeperDiscoveryImpl {
 
     /**
      * @param evtData Event data.
-     * @throws Exception If failed.
      */
-    private void notifyNodeFail(final ZkDiscoveryNodeFailEventData evtData) throws Exception
{
-        processNodeFail(evtData.failedNodeInternalId(), evtData.topologyVersion());
+    private void notifyNodeFail(final ZkDiscoveryNodeFailEventData evtData) {
+        notifyNodeFail(evtData.failedNodeInternalId(), evtData.topologyVersion());
     }
 
     /**
-     * @param nodeInternalId Failed node internal ID.
+     * @param nodeInternalOrder Node order.
      * @param topVer Topology version.
-     * @throws Exception If failed.
      */
-    private void processNodeFail(int nodeInternalId, long topVer) throws Exception {
-        final ZookeeperClusterNode failedNode = rtState.top.removeNode(nodeInternalId);
+    private void notifyNodeFail(int nodeInternalOrder, long topVer) {
+        final ZookeeperClusterNode failedNode = rtState.top.removeNode(nodeInternalOrder);
 
-        assert failedNode != null;
+        assert failedNode != null && !failedNode.isLocal() : failedNode;
 
-        if (failedNode.isLocal()) {
-            U.warn(log, "Received EVT_NODE_FAILED for local node.");
+        PingFuture pingFut = pingFuts.get(failedNode.order());
 
-            rtState.onCloseStart();
+        if (pingFut != null)
+            pingFut.onDone(false);
 
-            if (locNode.isClient() && clientReconnectEnabled) {
-                boolean reconnect = false;
+        final List<ClusterNode> topSnapshot = rtState.top.topologySnapshot();
 
-                synchronized (stateMux) {
-                    if (connState == ConnectionState.STARTED) {
-                        reconnect = true;
+        lsnr.onDiscovery(EVT_NODE_FAILED,
+            topVer,
+            failedNode,
+            topSnapshot,
+            Collections.<Long, Collection<ClusterNode>>emptyMap(),
+            null);
+    }
 
-                        connState = ConnectionState.DISCONNECTED;
-                    }
-                }
+    /**
+     * @param msg Message to log.
+     * @return Exception to be thrown.
+     */
+    private Exception localNodeFail(String msg) {
+        U.warn(log, msg);
+
+        rtState.onCloseStart();
 
-                if (reconnect) {
-                    UUID newId = UUID.randomUUID();
+        if (clientReconnectEnabled) {
+            assert locNode.isClient() : locNode;
 
-                    U.quietAndWarn(log, "Received EVT_NODE_FAILED for local node, will try
to reconnect with new id [" +
-                        "newId=" + newId +
-                        ", prevId=" + locNode.id() +
-                        ", locNode=" + locNode + ']');
+            boolean reconnect = false;
 
-                    runInWorkerThread(new ReconnectClosure(newId));
+            synchronized (stateMux) {
+                if (connState == ConnectionState.STARTED) {
+                    reconnect = true;
+
+                    connState = ConnectionState.DISCONNECTED;
                 }
             }
-            else
-                notifySegmented();
 
-            // Stop any further processing.
-            throw new ZookeeperClientFailedException("Received node failed event for local
node.");
-        }
-        else {
-            PingFuture pingFut = pingFuts.get(failedNode.order());
-
-            if (pingFut != null)
-                pingFut.onDone(false);
+            if (reconnect) {
+                UUID newId = UUID.randomUUID();
 
-            final List<ClusterNode> topSnapshot = rtState.top.topologySnapshot();
+                U.quietAndWarn(log, "Client node will try to reconnect with new id [" +
+                    "newId=" + newId +
+                    ", prevId=" + locNode.id() +
+                    ", locNode=" + locNode + ']');
 
-            lsnr.onDiscovery(EVT_NODE_FAILED,
-                topVer,
-                failedNode,
-                topSnapshot,
-                Collections.<Long, Collection<ClusterNode>>emptyMap(),
-                null);
+                runInWorkerThread(new ReconnectClosure(newId));
+            }
         }
+        else
+            notifySegmented();
+
+        // Stop any further processing.
+        return new ZookeeperClientFailedException(msg);
     }
 
     /**
@@ -2629,6 +2833,11 @@ public class ZookeeperDiscoveryImpl {
         if (zkClient != null)
             zkClient.close();
 
+        ZkCommunicationErrorProcessFuture commErrFut = commErrProcFut.get();
+
+        if (commErrFut != null)
+            commErrFut.onError(new IgniteCheckedException("Node stopped."));
+
         IgniteUtils.shutdownNow(ZookeeperDiscoveryImpl.class, utilityPool, log);
     }
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/9e4f2c20/modules/core/src/test/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoverySpiBasicTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoverySpiBasicTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoverySpiBasicTest.java
index cee2e76..a29f478 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoverySpiBasicTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoverySpiBasicTest.java
@@ -20,7 +20,9 @@ package org.apache.ignite.spi.discovery.zk.internal;
 import java.io.File;
 import java.io.Serializable;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.BitSet;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -62,6 +64,7 @@ import org.apache.ignite.internal.IgnitionEx;
 import org.apache.ignite.internal.managers.discovery.DiscoveryLocalJoinData;
 import org.apache.ignite.internal.managers.discovery.IgniteDiscoverySpi;
 import org.apache.ignite.internal.processors.cache.GridCacheAbstractFullApiSelfTest;
+import org.apache.ignite.internal.util.future.IgniteFinishedFutureImpl;
 import org.apache.ignite.internal.util.lang.GridAbsPredicate;
 import org.apache.ignite.internal.util.typedef.G;
 import org.apache.ignite.internal.util.typedef.internal.CU;
@@ -1712,6 +1715,8 @@ public class ZookeeperDiscoverySpiBasicTest extends GridCommonAbstractTest
{
     }
 
     /**
+     * TODO ZK
+     *
      * @throws Exception If failed.
      */
     public void _testCommunicationFailure() throws Exception {
@@ -1918,6 +1923,97 @@ public class ZookeeperDiscoverySpiBasicTest extends GridCommonAbstractTest
{
     /**
      * @throws Exception If failed.
      */
+    public void testCommunicationErrorResolve_KillNode_1() throws Exception {
+        communicationErrorResolve_KillNodes(2, Collections.singletonList(2L));
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testCommunicationErrorResolve_KillNode_2() throws Exception {
+        communicationErrorResolve_KillNodes(3, Collections.singletonList(2L));
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testCommunicationErrorResolve_KillNode_3() throws Exception {
+        communicationErrorResolve_KillNodes(10, Arrays.asList(2L, 4L, 6L));
+    }
+
+    /**
+     * @param startNodes Number of nodes to start.
+     * @param killNodes Nodes to kill by resolve process.
+     * @throws Exception If failed.
+     */
+    private void communicationErrorResolve_KillNodes(int startNodes, Collection<Long>
killNodes) throws Exception {
+        testCommSpi = true;
+
+        commProblemRslvr = new TestNodeKillCommunicationProblemResolver(killNodes);
+
+        startGrids(startNodes);
+
+        ZkTestCommunicationSpi commSpi = ZkTestCommunicationSpi.forNode(ignite(0));
+
+        commSpi.checkRes = new BitSet(startNodes);
+
+        ZookeeperDiscoverySpi spi = null;
+
+        for (Ignite node : G.allGrids()) {
+            ZookeeperDiscoverySpi spi0 = spi(node);
+
+            if (!killNodes.contains(node.cluster().localNode().order())) {
+                spi = spi0;
+
+                break;
+            }
+        }
+
+        assertNotNull(spi);
+
+        spi.onCommunicationConnectionError(ignite(1).cluster().localNode(), new Exception("test"));
+
+        int expNodes = startNodes - killNodes.size();
+
+        waitForTopology(expNodes);
+
+        for (Ignite node : G.allGrids())
+            assertFalse(killNodes.contains(node.cluster().localNode().order()));
+
+        startGrid(startNodes);
+
+        waitForTopology(expNodes + 1);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testCommunicationErrorResolve_KillCoordinator() throws Exception {
+        // Kill coordinator.
+        testCommSpi = true;
+
+        commProblemRslvr = new TestNodeKillCommunicationProblemResolver(Collections.singleton(1L));
+
+        startGrids(3);
+
+        ZkTestCommunicationSpi commSpi = ZkTestCommunicationSpi.forNode(ignite(2));
+
+        commSpi.checkRes = new BitSet(3);
+
+        ZookeeperDiscoverySpi spi = spi(ignite(1));
+
+        spi.onCommunicationConnectionError(ignite(0).cluster().localNode(), new Exception("test"));
+
+        waitForTopology(2);
+
+        startGrid(10);
+
+        waitForTopology(3);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
     public void testConnectionCheck() throws Exception {
        final int NODES = 5;
 
@@ -1928,9 +2024,7 @@ public class ZookeeperDiscoverySpiBasicTest extends GridCommonAbstractTest
{
 
            TcpCommunicationSpi spi = (TcpCommunicationSpi)node.configuration().getCommunicationSpi();
 
-           List<ClusterNode> nodes = new ArrayList<>();
-
-           nodes.addAll(node.cluster().nodes());
+           List<ClusterNode> nodes = new ArrayList<>(node.cluster().nodes());
 
            BitSet res = spi.checkConnection(nodes).get();
 
@@ -2622,10 +2716,40 @@ public class ZookeeperDiscoverySpiBasicTest extends GridCommonAbstractTest
{
     /**
      *
      */
+    static class TestNodeKillCommunicationProblemResolver implements CommunicationProblemResolver
{
+        /** */
+        final Collection<Long> killNodeOrders;
+
+        /**
+         * @param killOrders Killed nodes order.
+         */
+        TestNodeKillCommunicationProblemResolver(Collection<Long> killOrders) {
+            this.killNodeOrders = killOrders;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void resolve(CommunicationProblemContext ctx) {
+            List<ClusterNode> nodes = ctx.topologySnapshot();
+
+            assertTrue(nodes.size() > 0);
+
+            for (ClusterNode node : nodes) {
+                if (killNodeOrders.contains(node.order()))
+                    ctx.killNode(node);
+            }
+        }
+    }
+
+    /**
+     *
+     */
     static class ZkTestCommunicationSpi extends TcpCommunicationSpi {
         /** */
         private volatile CountDownLatch pingLatch;
 
+        /** */
+        private volatile BitSet checkRes;
+
         /**
          * @param ignite Node.
          * @return Node's communication SPI.
@@ -2646,6 +2770,14 @@ public class ZookeeperDiscoverySpiBasicTest extends GridCommonAbstractTest
{
                 throw new IgniteException(e);
             }
 
+            BitSet checkRes = this.checkRes;
+
+            if (checkRes != null) {
+                this.checkRes = null;
+
+                return new IgniteFinishedFutureImpl<>(checkRes);
+            }
+
             return super.checkConnection(nodes);
         }
     }


Mime
View raw message