ZOOKEEPER-3032: MAVEN MIGRATION - move java server, client

Author: Norbert Kalmar <nkalmar@yahoo.com>

Reviewers: hanm@apache.org, andor@apache.org

Closes #633 from nkalmar/ZOOKEEPER-3032 and squashes the following commits:

3f9a0eca [Norbert Kalmar] ZOOKEEPER-3032 fix flaky QuorumPeerMainTest.testLeaderElectionWithDisloyalVoter_stillHasMajority
27295ed6 [Norbert Kalmar] ZOOKEEPER-3032 - MAVEN MIGRATION - move java server, client

8 files changed, 1504 insertions, 1 deletions

diff --git a/zookeeper-client/zookeeper-client-c/tests/zkServer.sh b/zookeeper-client/zookeeper-client-c/tests/zkServer.sh
index 4496309e9..cd8d7bb80 100755
--- a/zookeeper-client/zookeeper-client-c/tests/zkServer.sh
+++ b/zookeeper-client/zookeeper-client-c/tests/zkServer.sh
@@ -90,7 +90,7 @@ do
     CLASSPATH="$CLASSPATH:$i"
 done
 
-for i in "${zk_base}"/src/java/lib/*.jar
+for i in "${zk_base}"/zookeeper-common/src/main/resource/lib/*.jar
 do
     CLASSPATH="$CLASSPATH:$i"
 done
diff --git a/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/ConnectStringParser.java b/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/ConnectStringParser.java
new file mode 100644
index 000000000..085e44dfb
--- /dev/null
+++ b/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/ConnectStringParser.java
@@ -0,0 +1,90 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.client;
+
+import org.apache.zookeeper.common.PathUtils;
+
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.zookeeper.common.StringUtils.split;
+
+/**
+ * A parser for ZooKeeper Client connect strings.
+ * 
+ * This class is not meant to be seen or used outside of ZooKeeper itself.
+ * 
+ * The chrootPath member should be replaced by a Path object in issue
+ * ZOOKEEPER-849.
+ * 
+ * @see org.apache.zookeeper.ZooKeeper
+ */
+public final class ConnectStringParser {
+    private static final int DEFAULT_PORT = 2181;
+
+    private final String chrootPath;
+
+    private final ArrayList<InetSocketAddress> serverAddresses = new ArrayList<InetSocketAddress>();
+
+    /**
+     * 
+     * @throws IllegalArgumentException
+     *             for an invalid chroot path.
+     */
+    public ConnectStringParser(String connectString) {
+        // parse out chroot, if any
+        int off = connectString.indexOf('/');
+        if (off >= 0) {
+            String chrootPath = connectString.substring(off);
+            // ignore "/" chroot spec, same as null
+            if (chrootPath.length() == 1) {
+                this.chrootPath = null;
+            } else {
+                PathUtils.validatePath(chrootPath);
+                this.chrootPath = chrootPath;
+            }
+            connectString = connectString.substring(0, off);
+        } else {
+            this.chrootPath = null;
+        }
+
+        List<String> hostsList = split(connectString,",");
+        for (String host : hostsList) {
+            int port = DEFAULT_PORT;
+            int pidx = host.lastIndexOf(':');
+            if (pidx >= 0) {
+                // otherwise : is at the end of the string, ignore
+                if (pidx < host.length() - 1) {
+                    port = Integer.parseInt(host.substring(pidx + 1));
+                }
+                host = host.substring(0, pidx);
+            }
+            serverAddresses.add(InetSocketAddress.createUnresolved(host, port));
+        }
+    }
+
+    public String getChrootPath() {
+        return chrootPath;
+    }
+
+    public ArrayList<InetSocketAddress> getServerAddresses() {
+        return serverAddresses;
+    }
+}
diff --git a/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/FourLetterWordMain.java b/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/FourLetterWordMain.java
new file mode 100644
index 000000000..b396c15c3
--- /dev/null
+++ b/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/FourLetterWordMain.java
@@ -0,0 +1,146 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.client;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocket;
+import javax.net.ssl.SSLSocketFactory;
+
+import org.apache.yetus.audience.InterfaceAudience;
+import org.apache.zookeeper.common.X509Exception.SSLContextException;
+import org.apache.zookeeper.common.X509Util;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@InterfaceAudience.Public
+public class FourLetterWordMain {
+    //in milliseconds, socket should connect/read within this period otherwise SocketTimeoutException
+    private static final int DEFAULT_SOCKET_TIMEOUT = 5000;
+    protected static final Logger LOG = LoggerFactory.getLogger(FourLetterWordMain.class);
+    /**
+     * Send the 4letterword
+     * @param host the destination host
+     * @param port the destination port
+     * @param cmd the 4letterword
+     * @return server response
+     * @throws java.io.IOException
+     * @throws SSLContextException
+     */
+    public static String send4LetterWord(String host, int port, String cmd)
+            throws IOException, SSLContextException {
+        return send4LetterWord(host, port, cmd, false, DEFAULT_SOCKET_TIMEOUT);
+    }
+
+    /**
+     * Send the 4letterword
+     * @param host the destination host
+     * @param port the destination port
+     * @param cmd the 4letterword
+     * @param secure whether to use SSL
+     * @return server response
+     * @throws java.io.IOException
+     * @throws SSLContextException
+     */
+    public static String send4LetterWord(String host, int port, String cmd, boolean secure)
+            throws IOException, SSLContextException {
+        return send4LetterWord(host, port, cmd, secure, DEFAULT_SOCKET_TIMEOUT);
+    }
+
+    /**
+     * Send the 4letterword
+     * @param host the destination host
+     * @param port the destination port
+     * @param cmd the 4letterword
+     * @param secure whether to use SSL
+     * @param timeout in milliseconds, maximum time to wait while connecting/reading data
+     * @return server response
+     * @throws java.io.IOException
+     * @throws SSLContextException
+     */
+    public static String send4LetterWord(String host, int port, String cmd, boolean secure, int timeout)
+            throws IOException, SSLContextException {
+        LOG.info("connecting to {} {}", host, port);
+        Socket sock;
+        InetSocketAddress hostaddress= host != null ? new InetSocketAddress(host, port) :
+            new InetSocketAddress(InetAddress.getByName(null), port);
+        if (secure) {
+            LOG.info("using secure socket");
+            SSLContext sslContext = X509Util.createSSLContext();
+            SSLSocketFactory socketFactory = sslContext.getSocketFactory();
+            SSLSocket sslSock = (SSLSocket) socketFactory.createSocket();
+            sslSock.connect(hostaddress, timeout);
+            sslSock.startHandshake();
+            sock = sslSock;
+        } else {
+            sock = new Socket();
+            sock.connect(hostaddress, timeout);
+        }
+        sock.setSoTimeout(timeout);
+        BufferedReader reader = null;
+        try {
+            OutputStream outstream = sock.getOutputStream();
+            outstream.write(cmd.getBytes());
+            outstream.flush();
+
+            // this replicates NC - close the output stream before reading
+            if (!secure) {
+                // SSL prohibits unilateral half-close
+                sock.shutdownOutput();
+            }
+
+            reader =
+                    new BufferedReader(
+                            new InputStreamReader(sock.getInputStream()));
+            StringBuilder sb = new StringBuilder();
+            String line;
+            while((line = reader.readLine()) != null) {
+                sb.append(line + "\n");
+            }
+            return sb.toString();
+        } catch (SocketTimeoutException e) {
+            throw new IOException("Exception while executing four letter word: " + cmd, e);
+        } finally {
+            sock.close();
+            if (reader != null) {
+                reader.close();
+            }
+        }
+    }
+    
+    public static void main(String[] args)
+            throws IOException, SSLContextException
+    {
+        if (args.length == 3) {
+            System.out.println(send4LetterWord(args[0], Integer.parseInt(args[1]), args[2]));
+        } else if (args.length == 4) {
+            System.out.println(send4LetterWord(args[0], Integer.parseInt(args[1]), args[2], Boolean.parseBoolean(args[3])));
+        } else {
+            System.out.println("Usage: FourLetterWordMain <host> <port> <cmd> <secure(optional)>");
+        }
+    }
+}
diff --git a/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/HostProvider.java b/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/HostProvider.java
new file mode 100644
index 000000000..caeedcc4e
--- /dev/null
+++ b/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/HostProvider.java
@@ -0,0 +1,76 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.client;
+
+import org.apache.yetus.audience.InterfaceAudience;
+
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.util.Collection;
+
+/**
+ * A set of hosts a ZooKeeper client should connect to.
+ * 
+ * Classes implementing this interface must guarantee the following:
+ * 
+ * * Every call to next() returns an InetSocketAddress. So the iterator never
+ * ends.
+ * 
+ * * The size() of a HostProvider may never be zero.
+ * 
+ * A HostProvider must return resolved InetSocketAddress instances on next() if the next address is resolvable.
+ * In that case, it's up to the HostProvider, whether it returns the next resolvable address in the list or return
+ * the next one as UnResolved.
+ * 
+ * Different HostProvider could be imagined:
+ * 
+ * * A HostProvider that loads the list of Hosts from an URL or from DNS 
+ * * A HostProvider that re-resolves the InetSocketAddress after a timeout. 
+ * * A HostProvider that prefers nearby hosts.
+ */
+@InterfaceAudience.Public
+public interface HostProvider {
+    public int size();
+
+    /**
+     * The next host to try to connect to.
+     * 
+     * For a spinDelay of 0 there should be no wait.
+     * 
+     * @param spinDelay
+     *            Milliseconds to wait if all hosts have been tried once.
+     */
+    public InetSocketAddress next(long spinDelay);
+
+    /**
+     * Notify the HostProvider of a successful connection.
+     * 
+     * The HostProvider may use this notification to reset it's inner state.
+     */
+    public void onConnected();
+
+    /**
+     * Update the list of servers. This returns true if changing connections is necessary for load-balancing, false otherwise.
+     * @param serverAddresses new host list
+     * @param currentHost the host to which this client is currently connected
+     * @return true if changing connections is necessary for load-balancing, false otherwise  
+     */
+    boolean updateServerList(Collection<InetSocketAddress> serverAddresses,
+        InetSocketAddress currentHost);
+}
diff --git a/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/StaticHostProvider.java b/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/StaticHostProvider.java
new file mode 100644
index 000000000..060210303
--- /dev/null
+++ b/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/StaticHostProvider.java
@@ -0,0 +1,383 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.client;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Most simple HostProvider, resolves on every next() call.
+ *
+ * Please be aware that although this class doesn't do any DNS caching, there're multiple levels of caching already
+ * present across the stack like in JVM, OS level, hardware, etc. The best we could do here is to get the most recent
+ * address from the underlying system which is considered up-to-date.
+ *
+ */
+@InterfaceAudience.Public
+public final class StaticHostProvider implements HostProvider {
+    public interface Resolver {
+        InetAddress[] getAllByName(String name) throws UnknownHostException;
+    }
+
+    private static final Logger LOG = LoggerFactory
+            .getLogger(StaticHostProvider.class);
+
+    private List<InetSocketAddress> serverAddresses = new ArrayList<InetSocketAddress>(
+            5);
+
+    private Random sourceOfRandomness;
+    private int lastIndex = -1;
+
+    private int currentIndex = -1;
+
+    /**
+     * The following fields are used to migrate clients during reconfiguration
+     */
+    private boolean reconfigMode = false;
+
+    private final List<InetSocketAddress> oldServers = new ArrayList<InetSocketAddress>(
+            5);
+
+    private final List<InetSocketAddress> newServers = new ArrayList<InetSocketAddress>(
+            5);
+
+    private int currentIndexOld = -1;
+    private int currentIndexNew = -1;
+
+    private float pOld, pNew;
+
+    private Resolver resolver;
+
+    /**
+     * Constructs a SimpleHostSet.
+     * 
+     * @param serverAddresses
+     *            possibly unresolved ZooKeeper server addresses
+     * @throws IllegalArgumentException
+     *             if serverAddresses is empty or resolves to an empty list
+     */
+    public StaticHostProvider(Collection<InetSocketAddress> serverAddresses) {
+        init(serverAddresses,
+                System.currentTimeMillis() ^ this.hashCode(),
+                new Resolver() {
+            @Override
+            public InetAddress[] getAllByName(String name) throws UnknownHostException {
+                return InetAddress.getAllByName(name);
+            }
+        });
+    }
+
+    /**
+     * Constructs a SimpleHostSet.
+     *
+     * Introduced for testing purposes. getAllByName() is a static method of InetAddress, therefore cannot be easily mocked.
+     * By abstraction of Resolver interface we can easily inject a mocked implementation in tests.
+     *
+     * @param serverAddresses
+     *              possibly unresolved ZooKeeper server addresses
+     * @param resolver
+     *              custom resolver implementation
+     */
+    public StaticHostProvider(Collection<InetSocketAddress> serverAddresses, Resolver resolver) {
+        init(serverAddresses, System.currentTimeMillis() ^ this.hashCode(), resolver);
+    }
+
+    /**
+     * Constructs a SimpleHostSet. This constructor is used from StaticHostProviderTest to produce deterministic test results
+     * by initializing sourceOfRandomness with the same seed
+     * 
+     * @param serverAddresses
+     *            possibly unresolved ZooKeeper server addresses
+     * @param randomnessSeed a seed used to initialize sourceOfRandomnes
+     * @throws IllegalArgumentException
+     *             if serverAddresses is empty or resolves to an empty list
+     */
+    public StaticHostProvider(Collection<InetSocketAddress> serverAddresses,
+        long randomnessSeed) {
+        init(serverAddresses, randomnessSeed, new Resolver() {
+            @Override
+            public InetAddress[] getAllByName(String name) throws UnknownHostException {
+                return InetAddress.getAllByName(name);
+            }
+        });
+    }
+
+    private void init(Collection<InetSocketAddress> serverAddresses, long randomnessSeed, Resolver resolver) {
+        this.sourceOfRandomness = new Random(randomnessSeed);
+        this.resolver = resolver;
+        if (serverAddresses.isEmpty()) {
+            throw new IllegalArgumentException(
+                    "A HostProvider may not be empty!");
+        }
+        this.serverAddresses = shuffle(serverAddresses);
+        currentIndex = -1;
+        lastIndex = -1;
+    }
+
+    private InetSocketAddress resolve(InetSocketAddress address) {
+        try {
+            String curHostString = address.getHostString();
+            List<InetAddress> resolvedAddresses = new ArrayList<>(Arrays.asList(this.resolver.getAllByName(curHostString)));
+            if (resolvedAddresses.isEmpty()) {
+                return address;
+            }
+            Collections.shuffle(resolvedAddresses);
+            return new InetSocketAddress(resolvedAddresses.get(0), address.getPort());
+        } catch (UnknownHostException e) {
+            LOG.error("Unable to resolve address: {}", address.toString(), e);
+            return address;
+        }
+    }
+
+    private List<InetSocketAddress> shuffle(Collection<InetSocketAddress> serverAddresses) {
+        List<InetSocketAddress> tmpList = new ArrayList<>(serverAddresses.size());
+        tmpList.addAll(serverAddresses);
+        Collections.shuffle(tmpList, sourceOfRandomness);
+        return tmpList;
+    }
+
+    /**
+     * Update the list of servers. This returns true if changing connections is necessary for load-balancing, false
+	 * otherwise. Changing connections is necessary if one of the following holds: 
+     * a) the host to which this client is currently connected is not in serverAddresses.
+     *    Otherwise (if currentHost is in the new list serverAddresses):   
+     * b) the number of servers in the cluster is increasing - in this case the load on currentHost should decrease,
+     *    which means that SOME of the clients connected to it will migrate to the new servers. The decision whether
+     *    this client migrates or not (i.e., whether true or false is returned) is probabilistic so that the expected 
+     *    number of clients connected to each server is the same.
+     *    
+     * If true is returned, the function sets pOld and pNew that correspond to the probability to migrate to ones of the
+     * new servers in serverAddresses or one of the old servers (migrating to one of the old servers is done only
+     * if our client's currentHost is not in serverAddresses). See nextHostInReconfigMode for the selection logic.
+     *
+     * See <a href="https://issues.apache.org/jira/browse/ZOOKEEPER-1355">ZOOKEEPER-1355</a>
+     * for the protocol and its evaluation, and StaticHostProviderTest for the tests that illustrate how load balancing
+     * works with this policy.
+     *
+     * @param serverAddresses new host list
+     * @param currentHost the host to which this client is currently connected
+     * @return true if changing connections is necessary for load-balancing, false otherwise  
+     */
+    @Override
+    public synchronized boolean updateServerList(
+            Collection<InetSocketAddress> serverAddresses,
+            InetSocketAddress currentHost) {
+        List<InetSocketAddress> shuffledList = shuffle(serverAddresses);
+        if (shuffledList.isEmpty()) {
+            throw new IllegalArgumentException(
+                    "A HostProvider may not be empty!");
+        }
+        // Check if client's current server is in the new list of servers
+        boolean myServerInNewConfig = false;
+
+        InetSocketAddress myServer = currentHost;
+
+        // choose "current" server according to the client rebalancing algorithm
+        if (reconfigMode) {
+            myServer = next(0);
+        }
+
+        // if the client is not currently connected to any server
+        if (myServer == null) {
+            // reconfigMode = false (next shouldn't return null).
+            if (lastIndex >= 0) {
+                // take the last server to which we were connected
+                myServer = this.serverAddresses.get(lastIndex);
+            } else {
+                // take the first server on the list
+                myServer = this.serverAddresses.get(0);
+            }
+        }
+
+        for (InetSocketAddress addr : shuffledList) {
+            if (addr.getPort() == myServer.getPort()
+                    && ((addr.getAddress() != null
+                            && myServer.getAddress() != null && addr
+                            .getAddress().equals(myServer.getAddress())) || addr
+                            .getHostString().equals(myServer.getHostString()))) {
+                myServerInNewConfig = true;
+                break;
+            }
+        }
+
+        reconfigMode = true;
+
+        newServers.clear();
+        oldServers.clear();
+        // Divide the new servers into oldServers that were in the previous list
+        // and newServers that were not in the previous list
+        for (InetSocketAddress address : shuffledList) {
+            if (this.serverAddresses.contains(address)) {
+                oldServers.add(address);
+            } else {
+                newServers.add(address);
+            }
+        }
+
+        int numOld = oldServers.size();
+        int numNew = newServers.size();
+
+        // number of servers increased
+        if (numOld + numNew > this.serverAddresses.size()) {
+            if (myServerInNewConfig) {
+                // my server is in new config, but load should be decreased.
+                // Need to decide if this client
+                // is moving to one of the new servers
+                if (sourceOfRandomness.nextFloat() <= (1 - ((float) this.serverAddresses
+                        .size()) / (numOld + numNew))) {
+                    pNew = 1;
+                    pOld = 0;
+                } else {
+                    // do nothing special - stay with the current server
+                    reconfigMode = false;
+                }
+            } else {
+                // my server is not in new config, and load on old servers must
+                // be decreased, so connect to
+                // one of the new servers
+                pNew = 1;
+                pOld = 0;
+            }
+        } else { // number of servers stayed the same or decreased
+            if (myServerInNewConfig) {
+                // my server is in new config, and load should be increased, so
+                // stay with this server and do nothing special
+                reconfigMode = false;
+            } else {
+                pOld = ((float) (numOld * (this.serverAddresses.size() - (numOld + numNew))))
+                        / ((numOld + numNew) * (this.serverAddresses.size() - numOld));
+                pNew = 1 - pOld;
+            }
+        }
+
+        if (!reconfigMode) {
+            currentIndex = shuffledList.indexOf(getServerAtCurrentIndex());
+        } else {
+            currentIndex = -1;
+        }
+        this.serverAddresses = shuffledList;
+        currentIndexOld = -1;
+        currentIndexNew = -1;
+        lastIndex = currentIndex;
+        return reconfigMode;
+    }
+
+    public synchronized InetSocketAddress getServerAtIndex(int i) {
+    	if (i < 0 || i >= serverAddresses.size()) return null;
+    	return serverAddresses.get(i);
+    }
+    
+    public synchronized InetSocketAddress getServerAtCurrentIndex() {
+    	return getServerAtIndex(currentIndex);
+    }
+
+    public synchronized int size() {
+        return serverAddresses.size();
+    }
+
+    /**
+     * Get the next server to connect to, when in "reconfigMode", which means that 
+     * you've just updated the server list, and now trying to find some server to connect to. 
+     * Once onConnected() is called, reconfigMode is set to false. Similarly, if we tried to connect
+     * to all servers in new config and failed, reconfigMode is set to false.
+     * 
+     * While in reconfigMode, we should connect to a server in newServers with probability pNew and to servers in
+     * oldServers with probability pOld (which is just 1-pNew). If we tried out all servers in either oldServers
+     * or newServers we continue to try servers from the other set, regardless of pNew or pOld. If we tried all servers
+     * we give up and go back to the normal round robin mode
+     *
+     * When called, this should be protected by synchronized(this)
+     */
+    private InetSocketAddress nextHostInReconfigMode() {
+        boolean takeNew = (sourceOfRandomness.nextFloat() <= pNew);
+
+        // take one of the new servers if it is possible (there are still such
+        // servers we didn't try),
+        // and either the probability tells us to connect to one of the new
+        // servers or if we already
+        // tried all the old servers
+        if (((currentIndexNew + 1) < newServers.size())
+                && (takeNew || (currentIndexOld + 1) >= oldServers.size())) {
+            ++currentIndexNew;
+            return newServers.get(currentIndexNew);
+        }
+
+        // start taking old servers
+        if ((currentIndexOld + 1) < oldServers.size()) {
+            ++currentIndexOld;
+            return oldServers.get(currentIndexOld);
+        }
+
+        return null;
+    }
+
+    public InetSocketAddress next(long spinDelay) {
+        boolean needToSleep = false;
+        InetSocketAddress addr;
+
+        synchronized(this) {
+            if (reconfigMode) {
+                addr = nextHostInReconfigMode();
+                if (addr != null) {
+                	currentIndex = serverAddresses.indexOf(addr);
+                	return resolve(addr);
+                }
+                //tried all servers and couldn't connect
+                reconfigMode = false;
+                needToSleep = (spinDelay > 0);
+            }        
+            ++currentIndex;
+            if (currentIndex == serverAddresses.size()) {
+                currentIndex = 0;
+            }            
+            addr = serverAddresses.get(currentIndex);
+            needToSleep = needToSleep || (currentIndex == lastIndex && spinDelay > 0);
+            if (lastIndex == -1) { 
+                // We don't want to sleep on the first ever connect attempt.
+                lastIndex = 0;
+            }
+        }
+        if (needToSleep) {
+            try {
+                Thread.sleep(spinDelay);
+            } catch (InterruptedException e) {
+                LOG.warn("Unexpected exception", e);
+            }
+        }
+
+        return resolve(addr);
+    }
+
+    public synchronized void onConnected() {
+        lastIndex = currentIndex;
+        reconfigMode = false;
+    }
+
+}
diff --git a/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/ZKClientConfig.java b/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/ZKClientConfig.java
new file mode 100644
index 000000000..097f2f0a5
--- /dev/null
+++ b/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/ZKClientConfig.java
@@ -0,0 +1,140 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.client;
+
+import java.io.File;
+
+import org.apache.yetus.audience.InterfaceAudience;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.common.ZKConfig;
+import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException;
+
+/**
+ * Handles client specific properties
+ * @since 3.5.2
+ */
+@InterfaceAudience.Public
+public class ZKClientConfig extends ZKConfig {
+    public static final String ZK_SASL_CLIENT_USERNAME = "zookeeper.sasl.client.username";
+    public static final String ZK_SASL_CLIENT_USERNAME_DEFAULT = "zookeeper";
+    @SuppressWarnings("deprecation")
+    public static final String LOGIN_CONTEXT_NAME_KEY = ZooKeeperSaslClient.LOGIN_CONTEXT_NAME_KEY;;
+    public static final String LOGIN_CONTEXT_NAME_KEY_DEFAULT = "Client";
+    @SuppressWarnings("deprecation")
+    public static final String ENABLE_CLIENT_SASL_KEY = ZooKeeperSaslClient.ENABLE_CLIENT_SASL_KEY;
+    @SuppressWarnings("deprecation")
+    public static final String ENABLE_CLIENT_SASL_DEFAULT = ZooKeeperSaslClient.ENABLE_CLIENT_SASL_DEFAULT;
+    public static final String ZOOKEEPER_SERVER_REALM = "zookeeper.server.realm";
+    /**
+     * This controls whether automatic watch resetting is enabled. Clients
+     * automatically reset watches during session reconnect, this option allows
+     * the client to turn off this behavior by setting the property
+     * "zookeeper.disableAutoWatchReset" to "true"
+     */
+    public static final String DISABLE_AUTO_WATCH_RESET = "zookeeper.disableAutoWatchReset";
+    @SuppressWarnings("deprecation")
+    public static final String ZOOKEEPER_CLIENT_CNXN_SOCKET = ZooKeeper.ZOOKEEPER_CLIENT_CNXN_SOCKET;
+    /**
+     * Setting this to "true" will enable encrypted client-server communication.
+     */
+    @SuppressWarnings("deprecation")
+    public static final String SECURE_CLIENT = ZooKeeper.SECURE_CLIENT;
+    public static final int CLIENT_MAX_PACKET_LENGTH_DEFAULT = 4096 * 1024; /* 4 MB */
+    public static final String ZOOKEEPER_REQUEST_TIMEOUT = "zookeeper.request.timeout";
+    /**
+     * Feature is disabled by default.
+     */
+    public static final long ZOOKEEPER_REQUEST_TIMEOUT_DEFAULT = 0;
+
+    public ZKClientConfig() {
+        super();
+        initFromJavaSystemProperties();
+    }
+
+    public ZKClientConfig(File configFile) throws ConfigException {
+        super(configFile);
+    }
+
+    public ZKClientConfig(String configPath) throws ConfigException {
+        super(configPath);
+    }
+
+    /**
+     * Initialize all the ZooKeeper client properties which are configurable as
+     * java system property
+     */
+    private void initFromJavaSystemProperties() {
+        setProperty(ZOOKEEPER_REQUEST_TIMEOUT,
+                System.getProperty(ZOOKEEPER_REQUEST_TIMEOUT));
+    }
+
+    @Override
+    protected void handleBackwardCompatibility() {
+        /**
+         * backward compatibility for properties which are common to both client
+         * and server
+         */
+        super.handleBackwardCompatibility();
+
+        /**
+         * backward compatibility for client specific properties
+         */
+        setProperty(ZK_SASL_CLIENT_USERNAME, System.getProperty(ZK_SASL_CLIENT_USERNAME));
+        setProperty(LOGIN_CONTEXT_NAME_KEY, System.getProperty(LOGIN_CONTEXT_NAME_KEY));
+        setProperty(ENABLE_CLIENT_SASL_KEY, System.getProperty(ENABLE_CLIENT_SASL_KEY));
+        setProperty(ZOOKEEPER_SERVER_REALM, System.getProperty(ZOOKEEPER_SERVER_REALM));
+        setProperty(DISABLE_AUTO_WATCH_RESET, System.getProperty(DISABLE_AUTO_WATCH_RESET));
+        setProperty(ZOOKEEPER_CLIENT_CNXN_SOCKET, System.getProperty(ZOOKEEPER_CLIENT_CNXN_SOCKET));
+        setProperty(SECURE_CLIENT, System.getProperty(SECURE_CLIENT));
+    }
+
+    /**
+     * Returns true if the SASL client is enabled. By default, the client is
+     * enabled but can be disabled by setting the system property
+     * <code>zookeeper.sasl.client</code> to <code>false</code>. See
+     * ZOOKEEPER-1657 for more information.
+     *
+     * @return true if the SASL client is enabled.
+     */
+    public boolean isSaslClientEnabled() {
+        return Boolean.valueOf(getProperty(ENABLE_CLIENT_SASL_KEY, ENABLE_CLIENT_SASL_DEFAULT));
+    }
+
+    /**
+     * Get the value of the <code>key</code> property as an <code>long</code>.
+     * If property is not set, the provided <code>defaultValue</code> is
+     * returned
+     *
+     * @param key
+     *            property key.
+     * @param defaultValue
+     *            default value.
+     * @throws NumberFormatException
+     *             when the value is invalid
+     * @return return property value as an <code>long</code>, or
+     *         <code>defaultValue</code>
+     */
+    public long getLong(String key, long defaultValue) {
+        String value = getProperty(key);
+        if (value != null) {
+            return Long.parseLong(value.trim());
+        }
+        return defaultValue;
+    }
+}
diff --git a/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/ZooKeeperSaslClient.java b/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/ZooKeeperSaslClient.java
new file mode 100644
index 000000000..855062629
--- /dev/null
+++ b/zookeeper-client/zookeeper-client-java/src/main/java/org/apache/zookeeper/client/ZooKeeperSaslClient.java
@@ -0,0 +1,477 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.client;
+
+import java.io.IOException;
+import java.security.PrivilegedActionException;
+import java.security.PrivilegedExceptionAction;
+
+import javax.security.auth.Subject;
+import javax.security.auth.login.AppConfigurationEntry;
+import javax.security.auth.login.Configuration;
+import javax.security.auth.login.LoginException;
+import javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslException;
+
+import org.apache.zookeeper.AsyncCallback;
+import org.apache.zookeeper.ClientCnxn;
+import org.apache.zookeeper.Login;
+import org.apache.zookeeper.SaslClientCallbackHandler;
+import org.apache.zookeeper.Watcher.Event.KeeperState;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.proto.GetSASLRequest;
+import org.apache.zookeeper.proto.SetSASLResponse;
+
+import org.apache.zookeeper.util.SecurityUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class manages SASL authentication for the client. It
+ * allows ClientCnxn to authenticate using SASL with a ZooKeeper server.
+ */
+public class ZooKeeperSaslClient {
+    /**
+     * @deprecated Use {@link ZKClientConfig#LOGIN_CONTEXT_NAME_KEY}
+     *             instead.
+     */
+    @Deprecated
+    public static final String LOGIN_CONTEXT_NAME_KEY = "zookeeper.sasl.clientconfig";
+    /**
+     * @deprecated Use {@link ZKClientConfig#ENABLE_CLIENT_SASL_KEY}
+     *             instead.
+     */
+    @Deprecated
+    public static final String ENABLE_CLIENT_SASL_KEY = "zookeeper.sasl.client";
+    /**
+     * @deprecated Use {@link ZKClientConfig#ENABLE_CLIENT_SASL_DEFAULT}
+     *             instead.
+     */
+    @Deprecated
+    public static final String ENABLE_CLIENT_SASL_DEFAULT = "true";
+    private volatile boolean initializedLogin = false; 
+
+    /**
+     * Returns true if the SASL client is enabled. By default, the client
+     * is enabled but can be disabled by setting the system property
+     * <code>zookeeper.sasl.client</code> to <code>false</code>. See
+     * ZOOKEEPER-1657 for more information.
+     *
+     * @return true if the SASL client is enabled.
+     * @deprecated Use {@link ZKClientConfig#isSaslClientEnabled} instead
+     */
+    @Deprecated
+    public static boolean isEnabled() {
+        return Boolean.valueOf(System.getProperty(
+                ZKClientConfig.ENABLE_CLIENT_SASL_KEY,
+                ZKClientConfig.ENABLE_CLIENT_SASL_DEFAULT));
+    }
+
+    private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperSaslClient.class);
+    private Login login = null;
+    private SaslClient saslClient;
+    private boolean isSASLConfigured = true;
+    private final ZKClientConfig clientConfig;
+
+    private byte[] saslToken = new byte[0];
+
+    public enum SaslState {
+        INITIAL,INTERMEDIATE,COMPLETE,FAILED
+    }
+
+    private SaslState saslState = SaslState.INITIAL;
+
+    private boolean gotLastPacket = false;
+    /** informational message indicating the current configuration status */
+    private final String configStatus;
+
+    public SaslState getSaslState() {
+        return saslState;
+    }
+
+    public String getLoginContext() {
+        if (login != null)
+            return login.getLoginContextName();
+        return null;
+    }
+
+    public ZooKeeperSaslClient(final String serverPrincipal, ZKClientConfig clientConfig) throws LoginException {
+        /**
+         * ZOOKEEPER-1373: allow system property to specify the JAAS
+         * configuration section that the zookeeper client should use.
+         * Default to "Client".
+         */
+        String clientSection = clientConfig.getProperty(
+                ZKClientConfig.LOGIN_CONTEXT_NAME_KEY,
+                ZKClientConfig.LOGIN_CONTEXT_NAME_KEY_DEFAULT);
+        this.clientConfig = clientConfig;
+        // Note that 'Configuration' here refers to javax.security.auth.login.Configuration.
+        AppConfigurationEntry entries[] = null;
+        RuntimeException runtimeException = null;
+        try {
+            entries = Configuration.getConfiguration()
+                    .getAppConfigurationEntry(clientSection);
+        } catch (SecurityException e) {
+            // handle below: might be harmless if the user doesn't intend to use JAAS authentication.
+            runtimeException = e;
+        } catch (IllegalArgumentException e) {
+            // third party customized getAppConfigurationEntry could throw IllegalArgumentException when JAAS
+            // configuration isn't set. We can reevaluate whether to catch RuntimeException instead when more 
+            // different types of RuntimeException found
+            runtimeException = e;
+        }
+        if (entries != null) {
+            this.configStatus = "Will attempt to SASL-authenticate using Login Context section '" + clientSection + "'";
+            this.saslClient = createSaslClient(serverPrincipal, clientSection);
+        } else {
+            // Handle situation of clientSection's being null: it might simply because the client does not intend to 
+            // use SASL, so not necessarily an error.
+            saslState = SaslState.FAILED;
+            String explicitClientSection = clientConfig
+                    .getProperty(ZKClientConfig.LOGIN_CONTEXT_NAME_KEY);
+            if (explicitClientSection != null) {
+                // If the user explicitly overrides the default Login Context, they probably expected SASL to
+                // succeed. But if we got here, SASL failed.
+                if (runtimeException != null) {
+                    throw new LoginException(
+                            "Zookeeper client cannot authenticate using the "
+                                    + explicitClientSection
+                                    + " section of the supplied JAAS configuration: '"
+                                    + clientConfig.getJaasConfKey() + "' because of a "
+                                    + "RuntimeException: " + runtimeException);
+                } else {
+                    throw new LoginException("Client cannot SASL-authenticate because the specified JAAS configuration " +
+                            "section '" + explicitClientSection + "' could not be found.");
+                }
+            } else {
+                // The user did not override the default context. It might be that they just don't intend to use SASL,
+                // so log at INFO, not WARN, since they don't expect any SASL-related information.
+                String msg = "Will not attempt to authenticate using SASL ";
+                if (runtimeException != null) {
+                    msg += "(" + runtimeException + ")";
+                } else {
+                    msg += "(unknown error)";
+                }
+                this.configStatus = msg;
+                this.isSASLConfigured = false;
+            }
+            if (clientConfig.getJaasConfKey() != null) {
+                // Again, the user explicitly set something SASL-related, so
+                // they probably expected SASL to succeed.
+                if (runtimeException != null) {
+                    throw new LoginException(
+                            "Zookeeper client cannot authenticate using the '"
+                                    + clientConfig.getProperty(
+                                            ZKClientConfig.LOGIN_CONTEXT_NAME_KEY,
+                                            ZKClientConfig.LOGIN_CONTEXT_NAME_KEY_DEFAULT)
+                                    + "' section of the supplied JAAS configuration: '"
+                                    + clientConfig.getJaasConfKey() + "' because of a "
+                                    + "RuntimeException: " + runtimeException);
+                } else {
+                    throw new LoginException(
+                            "No JAAS configuration section named '"
+                                    + clientConfig.getProperty(
+                                            ZKClientConfig.LOGIN_CONTEXT_NAME_KEY,
+                                            ZKClientConfig.LOGIN_CONTEXT_NAME_KEY_DEFAULT)
+                                    + "' was found in specified JAAS configuration file: '"
+                                    + clientConfig.getJaasConfKey() + "'.");
+                }
+            }
+        }
+    }
+    
+    /**
+     * @return informational message indicating the current configuration status.
+     */
+    public String getConfigStatus() {
+        return configStatus;
+    }
+
+    public boolean isComplete() {
+        return (saslState == SaslState.COMPLETE);
+    }
+
+    public boolean isFailed() {
+        return (saslState == SaslState.FAILED);
+    }
+
+    public static class ServerSaslResponseCallback implements AsyncCallback.DataCallback {
+        public void processResult(int rc, String path, Object ctx, byte data[], Stat stat) {
+            // processResult() is used by ClientCnxn's sendThread to respond to
+            // data[] contains the Zookeeper Server's SASL token.
+            // ctx is the ZooKeeperSaslClient object. We use this object's respondToServer() method
+            // to reply to the Zookeeper Server's SASL token
+            ZooKeeperSaslClient client = ((ClientCnxn)ctx).zooKeeperSaslClient;
+            if (client == null) {
+                LOG.warn("sasl client was unexpectedly null: cannot respond to Zookeeper server.");
+                return;
+            }
+            byte[] usedata = data;
+            if (data != null) {
+                LOG.debug("ServerSaslResponseCallback(): saslToken server response: (length="+usedata.length+")");
+            }
+            else {
+                usedata = new byte[0];
+                LOG.debug("ServerSaslResponseCallback(): using empty data[] as server response (length="+usedata.length+")");
+            }
+            client.respondToServer(usedata, (ClientCnxn)ctx);
+        }
+    }
+
+    private SaslClient createSaslClient(final String servicePrincipal, final String loginContext)
+            throws LoginException {
+        try {
+            if (!initializedLogin) {
+                synchronized (this) {
+                    if (login == null) {
+                        if (LOG.isDebugEnabled()) {
+                            LOG.debug("JAAS loginContext is: " + loginContext);
+                        }
+                        // note that the login object is static: it's shared amongst all zookeeper-related connections.
+                        // in order to ensure the login is initialized only once, it must be synchronized the code snippet.
+                        login = new Login(loginContext, new SaslClientCallbackHandler(null, "Client"), clientConfig);
+                        login.startThreadIfNeeded();
+                        initializedLogin = true;
+                    }
+                }
+            }
+            return SecurityUtils.createSaslClient(login.getSubject(),
+                    servicePrincipal, "zookeeper", "zk-sasl-md5", LOG, "Client");
+        } catch (LoginException e) {
+            // We throw LoginExceptions...
+            throw e;
+        } catch (Exception e) {
+            // ..but consume (with a log message) all other types of exceptions.
+            LOG.error("Exception while trying to create SASL client: " + e);
+            return null;
+        }
+    }
+
+    public void respondToServer(byte[] serverToken, ClientCnxn cnxn) {
+        if (saslClient == null) {
+            LOG.error("saslClient is unexpectedly null. Cannot respond to server's SASL message; ignoring.");
+            return;
+        }
+
+        if (!(saslClient.isComplete())) {
+            try {
+                saslToken = createSaslToken(serverToken);
+                if (saslToken != null) {
+                    sendSaslPacket(saslToken, cnxn);
+                }
+            } catch (SaslException e) {
+                LOG.error("SASL authentication failed using login context '" +
+                        this.getLoginContext() + "' with exception: {}", e);
+                saslState = SaslState.FAILED;
+                gotLastPacket = true;
+            }
+        }
+
+        if (saslClient.isComplete()) {
+            // GSSAPI: server sends a final packet after authentication succeeds
+            // or fails.
+            if ((serverToken == null) && (saslClient.getMechanismName().equals("GSSAPI")))
+                gotLastPacket = true;
+            // non-GSSAPI: no final packet from server.
+            if (!saslClient.getMechanismName().equals("GSSAPI")) {
+                gotLastPacket = true;
+            }
+            // SASL authentication is completed, successfully or not:
+            // enable the socket's writable flag so that any packets waiting for authentication to complete in
+            // the outgoing queue will be sent to the Zookeeper server.
+            cnxn.saslCompleted();
+        }
+    }
+
+    private byte[] createSaslToken() throws SaslException {
+        saslState = SaslState.INTERMEDIATE;
+        return createSaslToken(saslToken);
+    }
+
+    private byte[] createSaslToken(final byte[] saslToken) throws SaslException {
+        if (saslToken == null) {
+            // TODO: introspect about runtime environment (such as jaas.conf)
+            saslState = SaslState.FAILED;
+            throw new SaslException("Error in authenticating with a Zookeeper Quorum member: the quorum member's saslToken is null.");
+        }
+
+        Subject subject = login.getSubject();
+        if (subject != null) {
+            synchronized(login) {
+                try {
+                    final byte[] retval =
+                        Subject.doAs(subject, new PrivilegedExceptionAction<byte[]>() {
+                                public byte[] run() throws SaslException {
+                                    LOG.debug("saslClient.evaluateChallenge(len="+saslToken.length+")");
+                                    return saslClient.evaluateChallenge(saslToken);
+                                }
+                            });
+                    return retval;
+                }
+                catch (PrivilegedActionException e) {
+                    String error = "An error: (" + e + ") occurred when evaluating Zookeeper Quorum Member's " +
+                      " received SASL token.";
+                    // Try to provide hints to use about what went wrong so they can fix their configuration.
+                    // TODO: introspect about e: look for GSS information.
+                    final String UNKNOWN_SERVER_ERROR_TEXT =
+                      "(Mechanism level: Server not found in Kerberos database (7) - UNKNOWN_SERVER)";
+                    if (e.toString().indexOf(UNKNOWN_SERVER_ERROR_TEXT) > -1) {
+                        error += " This may be caused by Java's being unable to resolve the Zookeeper Quorum Member's" +
+                          " hostname correctly. You may want to try to adding" +
+                          " '-Dsun.net.spi.nameservice.provider.1=dns,sun' to your client's JVMFLAGS environment.";
+                    }
+                    error += " Zookeeper Client will go to AUTH_FAILED state.";
+                    LOG.error(error);
+                    saslState = SaslState.FAILED;
+                    throw new SaslException(error);
+                }
+            }
+        }
+        else {
+            throw new SaslException("Cannot make SASL token without subject defined. " +
+              "For diagnosis, please look for WARNs and ERRORs in your log related to the Login class.");
+        }
+    }
+
+    private void sendSaslPacket(byte[] saslToken, ClientCnxn cnxn)
+      throws SaslException{
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("ClientCnxn:sendSaslPacket:length="+saslToken.length);
+        }
+
+        GetSASLRequest request = new GetSASLRequest();
+        request.setToken(saslToken);
+        SetSASLResponse response = new SetSASLResponse();
+        ServerSaslResponseCallback cb = new ServerSaslResponseCallback();
+
+        try {
+            cnxn.sendPacket(request,response,cb, ZooDefs.OpCode.sasl);
+        } catch (IOException e) {
+            throw new SaslException("Failed to send SASL packet to server.",
+                e);
+        }
+    }
+
+    private void sendSaslPacket(ClientCnxn cnxn) throws SaslException {
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("ClientCnxn:sendSaslPacket:length="+saslToken.length);
+        }
+        GetSASLRequest request = new GetSASLRequest();
+        request.setToken(createSaslToken());
+        SetSASLResponse response = new SetSASLResponse();
+        ServerSaslResponseCallback cb = new ServerSaslResponseCallback();
+        try {
+            cnxn.sendPacket(request,response,cb, ZooDefs.OpCode.sasl);
+        } catch (IOException e) {
+            throw new SaslException("Failed to send SASL packet to server due " +
+              "to IOException:", e);
+        }
+    }
+
+    // used by ClientCnxn to know whether to emit a SASL-related event: either AuthFailed or SaslAuthenticated,
+    // or none, if not ready yet. Sets saslState to COMPLETE as a side-effect.
+    public KeeperState getKeeperState() {
+        if (saslClient != null) {
+            if (saslState == SaslState.FAILED) {
+              return KeeperState.AuthFailed;
+            }
+            if (saslClient.isComplete()) {
+                if (saslState == SaslState.INTERMEDIATE) {
+                    saslState = SaslState.COMPLETE;
+                    return KeeperState.SaslAuthenticated;
+                }
+            }
+        }
+        // No event ready to emit yet.
+        return null;
+    }
+
+    // Initialize the client's communications with the Zookeeper server by sending the server the first
+    // authentication packet.
+    public void initialize(ClientCnxn cnxn) throws SaslException {
+        if (saslClient == null) {
+            saslState = SaslState.FAILED;
+            throw new SaslException("saslClient failed to initialize properly: it's null.");
+        }
+        if (saslState == SaslState.INITIAL) {
+            if (saslClient.hasInitialResponse()) {
+                sendSaslPacket(cnxn);
+            }
+            else {
+                byte[] emptyToken = new byte[0];
+                sendSaslPacket(emptyToken, cnxn);
+            }
+            saslState = SaslState.INTERMEDIATE;
+        }
+    }
+
+    public boolean clientTunneledAuthenticationInProgress() {
+    	if (!isSASLConfigured) {
+    	    return false;
+        } 
+        // TODO: Rather than checking a disjunction here, should be a single member
+        // variable or method in this class to determine whether the client is
+        // configured to use SASL. (see also ZOOKEEPER-1455).
+        try {
+            if ((clientConfig.getJaasConfKey() != null)
+                    || ((Configuration.getConfiguration() != null) && (Configuration.getConfiguration()
+                            .getAppConfigurationEntry(clientConfig.getProperty(ZKClientConfig.LOGIN_CONTEXT_NAME_KEY,
+                                    ZKClientConfig.LOGIN_CONTEXT_NAME_KEY_DEFAULT)) != null))) {
+                // Client is configured to use a valid login Configuration, so
+                // authentication is either in progress, successful, or failed.
+
+                // 1. Authentication hasn't finished yet: we must wait for it to do so.
+                if ((isComplete() == false) &&
+                    (isFailed() == false)) {
+                    return true;
+                }
+
+                // 2. SASL authentication has succeeded or failed..
+                if (isComplete() || isFailed()) {
+                    if (gotLastPacket == false) {
+                        // ..but still in progress, because there is a final SASL
+                        // message from server which must be received.
+                    return true;
+                    }
+                }
+            }
+            // Either client is not configured to use a tunnelled authentication
+            // scheme, or tunnelled authentication has completed (successfully or
+            // not), and all server SASL messages have been received.
+            return false;
+        } catch (SecurityException e) {
+            // Thrown if the caller does not have permission to retrieve the Configuration.
+            // In this case, simply returning false is correct.
+            if (LOG.isDebugEnabled() == true) {
+                LOG.debug("Could not retrieve login configuration: " + e);
+            }
+            return false;
+        }
+    }
+
+    /**
+     * close login thread if running
+     */
+    public void shutdown() {
+        if (null != login) {
+            login.shutdown();
+        }
+    }
+}
diff --git a/zookeeper-client/zookeeper-client-java/src/test/java/org/apache/zookeeper/client/ZKClientConfigTest.java b/zookeeper-client/zookeeper-client-java/src/test/java/org/apache/zookeeper/client/ZKClientConfigTest.java
new file mode 100644
index 000000000..98f7c51bc
--- /dev/null
+++ b/zookeeper-client/zookeeper-client-java/src/test/java/org/apache/zookeeper/client/ZKClientConfigTest.java
@@ -0,0 +1,191 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.client;
+
+import static org.apache.zookeeper.client.ZKClientConfig.DISABLE_AUTO_WATCH_RESET;
+import static org.apache.zookeeper.client.ZKClientConfig.ENABLE_CLIENT_SASL_KEY;
+import static org.apache.zookeeper.client.ZKClientConfig.LOGIN_CONTEXT_NAME_KEY;
+import static org.apache.zookeeper.client.ZKClientConfig.SECURE_CLIENT;
+import static org.apache.zookeeper.client.ZKClientConfig.ZK_SASL_CLIENT_USERNAME;
+import static org.apache.zookeeper.client.ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET;
+import static org.apache.zookeeper.client.ZKClientConfig.ZOOKEEPER_SERVER_REALM;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.zookeeper.common.ZKConfig;
+import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.Timeout;
+
+public class ZKClientConfigTest {
+    private static final File testData = new File(System.getProperty("test.data.dir", "build/test/data"));
+    @Rule
+    public Timeout timeout = new Timeout(10, TimeUnit.SECONDS);
+
+    @BeforeClass
+    public static void init() {
+        if (!testData.exists()) {
+            testData.mkdirs();
+        }
+    }
+
+    @Test
+    public void testDefaultConfiguration() {
+        Map<String, String> properties = new HashMap<>();
+        properties.put(ZK_SASL_CLIENT_USERNAME, "zookeeper1");
+        properties.put(LOGIN_CONTEXT_NAME_KEY, "Client1");
+        properties.put(ENABLE_CLIENT_SASL_KEY, "true");
+        properties.put(ZOOKEEPER_SERVER_REALM, "zookeeper/hadoop.hadoop.com");
+        properties.put(DISABLE_AUTO_WATCH_RESET, "true");
+        properties.put(ZOOKEEPER_CLIENT_CNXN_SOCKET, "ClientCnxnSocketNetty");
+        properties.put(SECURE_CLIENT, "true");
+
+        for (Map.Entry<String, String> e : properties.entrySet()) {
+            System.setProperty(e.getKey(), e.getValue());
+        }
+        /**
+         * ZKClientConfig should get initialized with system properties
+         */
+        ZKClientConfig conf = new ZKClientConfig();
+        for (Map.Entry<String, String> e : properties.entrySet()) {
+            assertEquals(e.getValue(), conf.getProperty(e.getKey()));
+        }
+        /**
+         * clear properties
+         */
+        for (Map.Entry<String, String> e : properties.entrySet()) {
+            System.clearProperty(e.getKey());
+        }
+
+        conf = new ZKClientConfig();
+        /**
+         * test that all the properties are null
+         */
+        for (Map.Entry<String, String> e : properties.entrySet()) {
+            String result = conf.getProperty(e.getKey());
+            assertNull(result);
+        }
+    }
+
+    @Test
+    public void testSystemPropertyValue() {
+        String clientName = "zookeeper1";
+        System.setProperty(ZK_SASL_CLIENT_USERNAME, clientName);
+
+        ZKClientConfig conf = new ZKClientConfig();
+        assertEquals(conf.getProperty(ZK_SASL_CLIENT_USERNAME), clientName);
+
+        String newClientName = "zookeeper2";
+        conf.setProperty(ZK_SASL_CLIENT_USERNAME, newClientName);
+
+        assertEquals(conf.getProperty(ZK_SASL_CLIENT_USERNAME), newClientName);
+    }
+
+    @Test
+    public void testReadConfigurationFile() throws IOException, ConfigException {
+        File file = File.createTempFile("clientConfig", ".conf", testData);
+        file.deleteOnExit();
+        Properties clientConfProp = new Properties();
+        clientConfProp.setProperty(ENABLE_CLIENT_SASL_KEY, "true");
+        clientConfProp.setProperty(ZK_SASL_CLIENT_USERNAME, "ZK");
+        clientConfProp.setProperty(LOGIN_CONTEXT_NAME_KEY, "MyClient");
+        clientConfProp.setProperty(ZOOKEEPER_SERVER_REALM, "HADOOP.COM");
+        clientConfProp.setProperty("dummyProperty", "dummyValue");
+        OutputStream io = new FileOutputStream(file);
+        try {
+            clientConfProp.store(io, "Client Configurations");
+        } finally {
+            io.close();
+        }
+
+        ZKClientConfig conf = new ZKClientConfig();
+        conf.addConfiguration(file.getAbsolutePath());
+        assertEquals(conf.getProperty(ENABLE_CLIENT_SASL_KEY), "true");
+        assertEquals(conf.getProperty(ZK_SASL_CLIENT_USERNAME), "ZK");
+        assertEquals(conf.getProperty(LOGIN_CONTEXT_NAME_KEY), "MyClient");
+        assertEquals(conf.getProperty(ZOOKEEPER_SERVER_REALM), "HADOOP.COM");
+        assertEquals(conf.getProperty("dummyProperty"), "dummyValue");
+
+        // try to delete it now as we have done with the created file, why to
+        // wait for deleteOnExit() deletion
+        file.delete();
+
+    }
+
+    @Test
+    public void testSetConfiguration() {
+        ZKClientConfig conf = new ZKClientConfig();
+        String defaultValue = conf.getProperty(ZKClientConfig.ENABLE_CLIENT_SASL_KEY,
+                ZKClientConfig.ENABLE_CLIENT_SASL_DEFAULT);
+        if (defaultValue.equals("true")) {
+            conf.setProperty(ENABLE_CLIENT_SASL_KEY, "false");
+        } else {
+            conf.setProperty(ENABLE_CLIENT_SASL_KEY, "true");
+        }
+        assertTrue(conf.getProperty(ENABLE_CLIENT_SASL_KEY) != defaultValue);
+    }
+
+    @Test
+    public void testIntegerRetrievalFromProperty() {
+        ZKClientConfig conf = new ZKClientConfig();
+        String prop = "UnSetProperty" + System.currentTimeMillis();
+        int defaultValue = 100;
+        // property is not set we should get the default value
+        int result = conf.getInt(prop, defaultValue);
+        assertEquals(defaultValue, result);
+
+        // property is set but can not be parsed to int, we should get the
+        // NumberFormatException
+        conf.setProperty(ZKConfig.JUTE_MAXBUFFER, "InvlaidIntValue123");
+        try {
+            result = conf.getInt(ZKConfig.JUTE_MAXBUFFER, defaultValue);
+            fail("NumberFormatException is expected");
+        } catch (NumberFormatException exception) {
+            // do nothing
+        }
+        assertEquals(defaultValue, result);
+
+        // property is set to an valid int, we should get the set value
+        int value = ZKClientConfig.CLIENT_MAX_PACKET_LENGTH_DEFAULT;
+        conf.setProperty(ZKConfig.JUTE_MAXBUFFER, Integer.toString(value));
+        result = conf.getInt(ZKConfig.JUTE_MAXBUFFER, defaultValue);
+        assertEquals(value, result);
+
+        // property is set but with white spaces
+        value = 12345;
+        conf.setProperty(ZKConfig.JUTE_MAXBUFFER,
+                " " + Integer.toString(value) + " ");
+        result = conf.getInt(ZKConfig.JUTE_MAXBUFFER, defaultValue);
+        assertEquals(value, result);
+    }
+
+}