upgrade for new server of zookeeper: ZOOKEEPER-2795

Author: adyliu

README.md

@@ -4,24 +4,18 @@
 [![Build Status](https://travis-ci.org/adyliu/zkclient.png?branch=master)](https://travis-ci.org/adyliu/zkclient)
 
 
-* [2013-12-25] zkclient 2.1.1 released.
-* [2013-09-04] zkclient 2.1 released.
-* [2012-12-04] zkclient 2.0 released.
-* [2012-04-12] zkclient 1.0 released.
-
-
 ## Maven Dependency
 
-The version 2.1.1 has been synchronized to [Maven Central Repository](http://repo1.maven.org/maven2/com/github/adyliu/zkclient/).
+The latest version has been synchronized to [Maven Central Repository](http://repo1.maven.org/maven2/com/github/adyliu/zkclient/).
 
         <dependency>
             <groupId>com.github.adyliu</groupId>
             <artifactId>zkclient</artifactId>
-            <version>2.1.1</version>
+            <version>2.2.0</version>
         </dependency>
 
 
-* The version 2.x supports both zookeeper 3.3.x and 3.4.x.
+* The version 2.x supports both zookeeper 3.3.x and 3.4.x+.
 * The version [1.x](http://repo1.maven.org/maven2/com/github/adyliu/zkclient/) needs zookeeper 3.3.x.
 
 ## Wiki & Usage

pom.xml

@@ -5,7 +5,7 @@
     <groupId>com.github.adyliu</groupId>
     <artifactId>zkclient</artifactId>
     <name>zookeeper client wrapper</name>
-    <version>2.1.2</version>
+    <version>2.2.0</version>
     <packaging>jar</packaging>
     <description>A zookeeper client wrapper.</description>
     <url>https://github.com/adyliu/zkclient</url>
@@ -45,7 +45,7 @@
         <dependency>
             <groupId>org.apache.zookeeper</groupId>
             <artifactId>zookeeper</artifactId>
-            <version>3.4.9</version>
+            <version>[3.4.9,3.4.14]</version>
             <exclusions>
                 <exclusion>
                     <artifactId>jline</artifactId>

src/main/java/com/github/zkclient/ZkServer.java

@@ -1,12 +1,12 @@
 /**
  * Copyright 2010 the original author or authors.
- *
+ * <p>
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
- *
+ * <p>
  * http://www.apache.org/licenses/LICENSE-2.0
- *
+ * <p>
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -15,22 +15,22 @@
  */
 package com.github.zkclient;
 
-import java.io.File;
-import java.io.IOException;
-
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
-
-import org.apache.zookeeper.server.ZooKeeperServer;
-
 import com.github.zkclient.exception.ZkException;
-import com.github.zkclient.exception.ZkInterruptedException;
+import org.apache.zookeeper.client.FourLetterWordMain;
+import org.apache.zookeeper.server.ServerConfig;
+import org.apache.zookeeper.server.ZooKeeperServerMain;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class ZkServer {
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.io.File;
+import java.net.ConnectException;
 
-    private final static Logger LOG = LoggerFactory.getLogger(ZkServer.class);;
+public class ZkServer extends ZooKeeperServerMain {
+
+    private final static Logger LOG = LoggerFactory.getLogger(ZkServer.class);
+    ;
 
     public static final int DEFAULT_PORT = 2181;
 
@@ -42,10 +42,6 @@ public class ZkServer {
 
     private final String _logDir;
 
-    private ZooKeeperServer _zk;
-
-    private ServerCnxnFactory _nioFactory;
-
     private ZkClient _zkClient;
 
     private final int _port;
@@ -54,6 +50,10 @@ public class ZkServer {
 
     private final int _minSessionTimeout;
 
+    private volatile boolean shutdown = false;
+
+    private boolean daemon = true;
+
     public ZkServer(String dataDir, String logDir) {
         this(dataDir, logDir, DEFAULT_PORT);
     }
@@ -80,6 +80,7 @@ public int getPort() {
 
     @PostConstruct
     public void start() {
+        shutdown = false;
         startZkServer();
         _zkClient = new ZkClient("localhost:" + _port, 10000);
     }
@@ -90,68 +91,114 @@ private void startZkServer() {
             final File dataDir = new File(_dataDir);
             final File dataLogDir = new File(_logDir);
             dataDir.mkdirs();
-            dataLogDir.mkdirs();
 
             // single zk server
-            LOG.info("Start single zookeeper server...");
-            LOG.info("data dir: " + dataDir.getAbsolutePath());
-            LOG.info("data log dir: " + dataLogDir.getAbsolutePath());
-            startSingleZkServer(_tickTime, dataDir, dataLogDir, port);
+            LOG.info("Start single zookeeper server, port={} data={} ", port, dataDir.getAbsolutePath());
+            //
+            final ZooKeeperServerMain serverMain = this;
+            final InnerServerConfig config = new InnerServerConfig();
+            config.parse(new String[]{"" + port, dataDir.getAbsolutePath(), "" + _tickTime, "60"});
+            config.setMinSessionTimeout(_minSessionTimeout);
+            //
+            final String threadName = "inner-zkserver-" + port;
+            final Thread innerThread = new Thread(new Runnable() {
+                @Override
+                public void run() {
+                    try {
+                        serverMain.runFromConfig(config);
+                    } catch (Exception e) {
+                        throw new ZkException("Unable to start single ZooKeeper server.", e);
+                    }
+                }
+            }, threadName);
+            innerThread.setDaemon(daemon);
+            innerThread.start();
+            //
+            waitForServerUp(port, 30000, false);
+
         } else {
             throw new IllegalStateException("Zookeeper port " + port + " was already in use. Running in single machine mode?");
         }
     }
 
-    private void startSingleZkServer(final int tickTime, final File dataDir, final File dataLogDir, final int port) {
-        try {
-            _zk = new ZooKeeperServer(dataDir, dataLogDir, tickTime);
-            _zk.setMinSessionTimeout(_minSessionTimeout);
-            _nioFactory = ServerCnxnFactory.createFactory(port, 60);
-            _nioFactory.startup(_zk);
-        } catch (IOException e) {
-            throw new ZkException("Unable to start single ZooKeeper server.", e);
-        } catch (InterruptedException e) {
-            throw new ZkInterruptedException(e);
-        }
-    }
-
     @PreDestroy
     public void shutdown() {
-        ZooKeeperServer zk = _zk;
-        if (zk == null) {
-            LOG.warn("shutdown duplication");
-            return;
-        }else {
-            _zk = null;
+        if (!shutdown) {
+            shutdown = true;
+            LOG.info("Shutting down ZkServer port={}...", _port);
+            if (_zkClient != null) {
+                try {
+                    _zkClient.close();
+                } catch (ZkException e) {
+                    LOG.warn("Error on closing zkclient: " + e.getClass().getName());
+                }
+                _zkClient = null;
+            }
+            super.shutdown();
+            waitForServerDown(_port, 30000, false);
+            LOG.info("Shutting down ZkServer port={}...done", _port);
         }
-        LOG.info("Shutting down ZkServer...");
-        try {
-            _zkClient.close();
-        } catch (ZkException e) {
-            LOG.warn("Error on closing zkclient: " + e.getClass().getName());
+    }
+
+
+    public ZkClient getZkClient() {
+        return _zkClient;
+    }
+
+    class InnerServerConfig extends ServerConfig {
+        public void setMinSessionTimeout(int minSessionTimeout) {
+            this.minSessionTimeout = minSessionTimeout;
         }
-        if (_nioFactory != null) {
-            _nioFactory.shutdown();
+    }
+
+    public static boolean waitForServerUp(int port, long timeout, boolean secure) {
+        long start = System.currentTimeMillis();
+        while (true) {
             try {
-                _nioFactory.join();
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
+                // if there are multiple hostports, just take the first one
+                String result = FourLetterWordMain.send4LetterWord("127.0.0.1", port, "stat");
+                if (result.startsWith("Zookeeper version:") &&
+                        !result.contains("READ-ONLY")) {
+                    return true;
+                }
+            } catch (ConnectException e) {
+                // ignore as this is expected, do not log stacktrace
+                LOG.debug("server {} not up: {}", port, e.toString());
+            } catch (Exception e) {
+                // ignore as this is expected
+                LOG.info("server {} not up", port, e);
+            }
+
+            if (System.currentTimeMillis() > start + timeout) {
+                break;
             }
-            _nioFactory = null;
-        }
-        zk.shutdown();
-        if (zk.getZKDatabase() != null) {
             try {
-                // release file description
-                zk.getZKDatabase().close();
-            } catch (IOException e) {
-                LOG.error(e.getMessage(), e);
+                Thread.sleep(250);
+            } catch (InterruptedException e) {
+                // ignore
             }
         }
-        LOG.info("Shutting down ZkServer...done");
+        return false;
     }
 
-    public ZkClient getZkClient() {
-        return _zkClient;
+    public static boolean waitForServerDown(int port, long timeout, boolean secure) {
+        long start = System.currentTimeMillis();
+        while (true) {
+            try {
+                FourLetterWordMain.send4LetterWord("127.0.0.1", port, "stat");
+            } catch (Exception e) {
+                return true;
+            }
+
+            if (System.currentTimeMillis() > start + timeout) {
+                break;
+            }
+            try {
+                Thread.sleep(250);
+            } catch (InterruptedException e) {
+                // ignore
+            }
+        }
+        return false;
     }
 }

src/test/java/com/github/zkclient/ZkClientTest.java

@@ -79,7 +79,7 @@ private static ZkServer startZkServer(String testName, int port) throws IOExcept
 
     @AfterClass
     public static void cleanup() throws IOException {
-        deleteFile(new File(".", "build/test").getCanonicalFile());
+       deleteFile(new File(".", "build/test").getCanonicalFile());
     }
 
     @Before

src/test/java/com/github/zkclient/ZkServerDemo.java

@@ -0,0 +1,21 @@
+package com.github.zkclient;
+
+import java.io.File;
+import java.util.Arrays;
+
+public class ZkServerDemo {
+    public static void main(String[] args) throws  Exception{
+        File dataDir = new File("/tmp/zkdemo");
+        dataDir.mkdirs();
+        ZkServer server = new ZkServer(dataDir.getAbsolutePath(), dataDir.getAbsolutePath());
+        server.start();
+
+        ZkClient client =  server.getZkClient();
+        client.createPersistent("/a", true);
+        byte[] dat = client.readData("/a");
+        System.out.println(Arrays.toString(dat));
+        client.writeData("/a", "OK".getBytes());
+        System.out.println("agian="+Arrays.toString(dat));
+        //server.shutdown();
+    }
+}