ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject ignite git commit: zk
Date Thu, 23 Nov 2017 13:12:12 GMT
Repository: ignite
Updated Branches:
  refs/heads/ignite-zk f76526396 -> ce7f0cbcf


zk


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

Branch: refs/heads/ignite-zk
Commit: ce7f0cbcff948852cca4827c42ff021a6c7c62be
Parents: f765263
Author: sboikov <sboikov@gridgain.com>
Authored: Thu Nov 23 16:00:10 2017 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Thu Nov 23 16:12:06 2017 +0300

----------------------------------------------------------------------
 .../discovery/zk/internal/ZookeeperClient.java  | 80 ++++++++++++++++++--
 .../zk/internal/ZookeeperDiscoveryImpl.java     | 19 +++--
 .../zk/internal/ZookeeperClientTest.java        | 22 ++++++
 .../ZookeeperDiscoverySpiBasicTest.java         |  4 +
 4 files changed, 113 insertions(+), 12 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/ce7f0cbc/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 d4d23ee..9125982 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
@@ -29,7 +29,6 @@ import org.apache.zookeeper.AsyncCallback;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.Op;
-import org.apache.zookeeper.OpResult;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.ZooDefs;
@@ -206,6 +205,12 @@ public class ZookeeperClient implements Watcher {
             connLostC.run();
     }
 
+    /**
+     * @param path Path.
+     * @return {@code True} if node exists.
+     * @throws ZookeeperClientFailedException If connection to zk was lost.
+     * @throws InterruptedException If interrupted.
+     */
     boolean exists(String path) throws ZookeeperClientFailedException, InterruptedException
{
         for (;;) {
             long connStartTime = this.connStartTime;
@@ -219,8 +224,16 @@ public class ZookeeperClient implements Watcher {
         }
     }
 
-    void createAllIfNeeded(List<String> paths, CreateMode createMode)
-        throws ZookeeperClientFailedException, InterruptedException
+    /**
+     *
+     * @param paths Paths to create.
+     * @param createMode Create mode.
+     * @throws KeeperException.NodeExistsException If at least one of target node already
exists.
+     * @throws ZookeeperClientFailedException If connection to zk was lost.
+     * @throws InterruptedException If interrupted.
+     */
+    void createAll(List<String> paths, CreateMode createMode)
+        throws ZookeeperClientFailedException, InterruptedException, KeeperException.NodeExistsException
     {
         // TODO ZK: need check for max size?
         List<Op> ops = new ArrayList<>(paths.size());
@@ -282,7 +295,7 @@ public class ZookeeperClient implements Watcher {
             long connStartTime = this.connStartTime;
 
             try {
-                if (first) {
+                if (!first) {
                     List<String> children = zk.getChildren(dir, false);
 
                     for (int i = 0; i < children.size(); i++) {
@@ -315,6 +328,12 @@ public class ZookeeperClient implements Watcher {
         }
     }
 
+    /**
+     * @param path Path.
+     * @return Children nodes.
+     * @throws ZookeeperClientFailedException If connection to zk was lost.
+     * @throws InterruptedException If interrupted.
+     */
     List<String> getChildren(String path)
         throws ZookeeperClientFailedException, InterruptedException
     {
@@ -371,6 +390,13 @@ public class ZookeeperClient implements Watcher {
         }
     }
 
+    /**
+     * @param path Path.
+     * @param ver Version.
+     * @throws KeeperException.NoNodeException If target node does not exist.
+     * @throws ZookeeperClientFailedException If connection to zk was lost.
+     * @throws InterruptedException If interrupted.
+     */
     void delete(String path, int ver)
         throws KeeperException.NoNodeException, ZookeeperClientFailedException, InterruptedException
     {
@@ -391,6 +417,13 @@ public class ZookeeperClient implements Watcher {
         }
     }
 
+    /**
+     * @param path Path.
+     * @param data Data.
+     * @param ver Version.
+     * @throws ZookeeperClientFailedException If connection to zk was lost.
+     * @throws InterruptedException If interrupted.
+     */
     void setData(String path, byte[] data, int ver)
         throws ZookeeperClientFailedException, InterruptedException
     {
@@ -411,6 +444,13 @@ public class ZookeeperClient implements Watcher {
         }
     }
 
+    /**
+     * @param path Path.
+     * @return Data.
+     * @throws KeeperException.NoNodeException If target node does not exist.
+     * @throws ZookeeperClientFailedException If connection to zk was lost.
+     * @throws InterruptedException If interrupted.
+     */
     byte[] getData(String path)
         throws KeeperException.NoNodeException, ZookeeperClientFailedException, InterruptedException
     {
@@ -429,24 +469,45 @@ public class ZookeeperClient implements Watcher {
         }
     }
 
+    /**
+     * @param path Path.
+     * @param watcher Watcher.
+     * @param cb Callback.
+     */
     void existsAsync(String path, Watcher watcher, AsyncCallback.StatCallback cb) {
         ExistsOperation op = new ExistsOperation(path, watcher, cb);
 
         zk.exists(path, watcher, new StatCallbackWrapper(op), null);
     }
 
+    /**
+     * @param path Path.
+     * @param watcher Watcher.
+     * @param cb Callback.
+     */
     void getChildrenAsync(String path, Watcher watcher, AsyncCallback.Children2Callback cb)
{
         GetChildrenOperation op = new GetChildrenOperation(path, watcher, cb);
 
         zk.getChildren(path, watcher, new ChildrenCallbackWrapper(op), null);
     }
 
+    /**
+     * @param path Path.
+     * @param watcher Watcher.
+     * @param cb Callback.
+     */
     void getDataAsync(String path, Watcher watcher, AsyncCallback.DataCallback cb) {
         GetDataOperation op = new GetDataOperation(path, watcher, cb);
 
         zk.getData(path, watcher, new DataCallbackWrapper(op), null);
     }
 
+    /**
+     * @param path Path.
+     * @param data Data.
+     * @param createMode Create mode.
+     * @param cb Callback.
+     */
     void createAsync(String path, byte[] data, CreateMode createMode, AsyncCallback.StringCallback
cb) {
         if (data == null)
             data = EMPTY_BYTES;
@@ -547,8 +608,9 @@ public class ZookeeperClient implements Watcher {
      * @return {@code True} if can retry operation.
      */
     private boolean needRetry(int code) {
-        // TODO ZL: other codes.
-        return code == KeeperException.Code.CONNECTIONLOSS.intValue();
+        return code == KeeperException.Code.CONNECTIONLOSS.intValue() ||
+            code == KeeperException.Code.SESSIONMOVED.intValue() ||
+            code == KeeperException.Code.OPERATIONTIMEOUT.intValue();
     }
 
     /**
@@ -685,6 +747,12 @@ public class ZookeeperClient implements Watcher {
         /** */
         private final AsyncCallback.StringCallback cb;
 
+        /**
+         * @param path path.
+         * @param data Data.
+         * @param createMode Create mode.
+         * @param cb Callback.
+         */
         CreateOperation(String path, byte[] data, CreateMode createMode, AsyncCallback.StringCallback
cb) {
             this.path = path;
             this.data = data;

http://git-wip-us.apache.org/repos/asf/ignite/blob/ce7f0cbc/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 1cfa2eb..f9786d6 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
@@ -333,7 +333,6 @@ public class ZookeeperDiscoveryImpl {
      */
     private void initZkNodes() throws InterruptedException {
         try {
-            // TODO ZK: use multi.
             if (zkClient.exists(zkPaths.aliveNodesDir))
                 return; // This path is created last, assume all others dirs are created.
 
@@ -347,7 +346,15 @@ public class ZookeeperDiscoveryImpl {
             dirs.add(zkPaths.customEvtsAcksDir);
             dirs.add(zkPaths.aliveNodesDir);
 
-            zkClient.createAllIfNeeded(dirs, PERSISTENT);
+            try {
+                zkClient.createAll(dirs, PERSISTENT);
+            }
+            catch (KeeperException.NodeExistsException e) {
+                U.warn(log, "Failed to create nodes using bulk operation: " + e);
+
+                for (String dir : dirs)
+                    zkClient.createIfNeeded(dir, null, PERSISTENT);
+            }
         }
         catch (ZookeeperClientFailedException e) {
             throw new IgniteSpiException("Failed to initialize Zookeeper nodes", e);
@@ -656,14 +663,14 @@ public class ZookeeperDiscoveryImpl {
         boolean newEvts = false;
 
         for (String child : aliveNodes) {
-            Integer inernalId = ZkIgnitePaths.aliveInternalId(child);
+            Integer internalId = ZkIgnitePaths.aliveInternalId(child);
 
-            Object old = alives.put(inernalId, child);
+            Object old = alives.put(internalId, child);
 
             assert old == null;
 
-            if (!top.nodesByInternalId.containsKey(inernalId)) {
-                generateNodeJoin(curTop, inernalId, child);
+            if (!top.nodesByInternalId.containsKey(internalId)) {
+                generateNodeJoin(curTop, internalId, child);
 
                 watchAliveNodeData(child);
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/ce7f0cbc/modules/core/src/test/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperClientTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperClientTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperClientTest.java
index 8aac456..81edcde 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperClientTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperClientTest.java
@@ -17,6 +17,7 @@
 
 package org.apache.ignite.spi.discovery.zk.internal;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.Callable;
@@ -52,6 +53,27 @@ public class ZookeeperClientTest extends GridCommonAbstractTest {
     /**
      * @throws Exception If failed.
      */
+    public void testCreateAll() throws Exception {
+        startZK(1);
+
+        ZookeeperClient client = new ZookeeperClient(log, zkCluster.getConnectString(), 3000,
null);
+
+        client.createIfNeeded("/apacheIgnite", null, CreateMode.PERSISTENT);
+
+        List<String> paths = new ArrayList<>();
+
+        paths.add("/apacheIgnite/1");
+        paths.add("/apacheIgnite/2");
+        paths.add("/apacheIgnite/3");
+
+        client.createAll(paths, CreateMode.PERSISTENT);
+
+        assertEquals(3, client.getChildren("/apacheIgnite").size());
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
     public void testDeleteAll() throws Exception {
         startZK(1);
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/ce7f0cbc/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 f20cb19..9bdf670 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
@@ -870,6 +870,8 @@ public class ZookeeperDiscoverySpiBasicTest extends GridCommonAbstractTest
{
      * @throws Exception If failed.
      */
     private void topologyChangeWithRestarts(boolean restartZk, boolean closeClientSock) throws
Exception {
+        sesTimeout = 30_000;
+
         if (closeClientSock)
             testSockNio = true;
 
@@ -986,6 +988,8 @@ public class ZookeeperDiscoverySpiBasicTest extends GridCommonAbstractTest
{
      * @throws Exception If failed.
      */
     private void randomTopologyChanges(boolean restartZk, boolean closeClientSock) throws
Exception {
+        sesTimeout = 30_000;
+
         if (closeClientSock)
             testSockNio = true;
 


Mime
View raw message