ZOOKEEPER-3874: Official API to start ZooKeeper server from Java

Create an Official API to start a ZooKeeper server node from Java code.
The idea is not to run ZooKeeper server inside the same process of an application, but only to have a standard Launcher that can be used from Java and not a bash script.

See more context here
https://issues.apache.org/jira/browse/ZOOKEEPER-3874

This is how a Java launcher will look like for tests:
```
        int clientPort = PortAssignment.unique();
        final Properties configZookeeper = new Properties();
        configZookeeper.put("clientPort", clientPort + "");
        configZookeeper.put("host", "localhost");
        configZookeeper.put("..........................");
        try (ZooKeeperServerEmbedded zkServer = ZooKeeperServerEmbedded
                .builder()
                .baseDir(baseDir)
                .configuration(configZookeeper)
                .exitHandler(ExitHandler.LOG_ONLY)
                .build()) {
            zkServer.start();
            //// wait.....
      }
```

This feature does not overlap with Curator TestingServer, this feature is meant to be used a fundation for projects like TestingServer but also to run ZooKeeper server nodes in production.

This code is running in production at https://www.mag-news.com and https://emailsuccess.com, in such products we are using a Java based process manager

Author: Enrico Olivelli <eolivelli@apache.org>
Author: Enrico Olivelli <eolivelli@gmail.com>

Reviewers: Damien Diederen <ddiederen@apache.org>

Closes #1526 from eolivelli/fix/ZOOKEEPER-3874-embedded-api

.gitignore

@@ -71,7 +71,6 @@ zookeeper-server/src/main/resources/lib/ant-eclipse-*
 zookeeper-server/src/main/resources/lib/ivy-*
 zookeeper-server/src/main/java/org/apache/zookeeper/version/Info.java
 zookeeper-server/src/main/java/org/apache/zookeeper/version/VersionInfoMain.java
-zookeeper-server/src/test/resources/
 zookeeper-client/zookeeper-client-c/Makefile.in
 zookeeper-client/zookeeper-client-c/aclocal.m4
 zookeeper-client/zookeeper-client-c/autom4te.cache/

pom.xml

@@ -425,6 +425,7 @@
     <surefire.version>2.22.1</surefire.version>
 
     <surefire-forkcount>8</surefire-forkcount>
+    <redirectTestOutputToFile>true</redirectTestOutputToFile>
 
     <!-- dependency versions -->
     <slf4j.version>1.7.30</slf4j.version>
@@ -702,6 +703,10 @@
         <plugin>
           <groupId>org.apache.maven.plugins</groupId>
           <artifactId>maven-surefire-plugin</artifactId>
+          <configuration>
+              <trimStackTrace>false</trimStackTrace>
+              <redirectTestOutputToFile>${redirectTestOutputToFile}</redirectTestOutputToFile>
+          </configuration>
         </plugin>
         <plugin>
           <groupId>org.apache.maven.plugins</groupId>
@@ -961,6 +966,7 @@
             <exclude>src/main/resources/markdown/skin/*</exclude>
             <exclude>src/main/resources/markdown/html/*</exclude>
             <exclude>src/main/resources/markdown/images/*</exclude>
+            <exclude>**/src/test/resources/embedded/*.conf</exclude>
             <!-- contrib -->
             <exclude>**/JMX-RESOURCES</exclude>
             <exclude>**/src/main/resources/mainClasses</exclude>

zookeeper-server/pom.xml

@@ -265,7 +265,6 @@
           <reuseForks>false</reuseForks>
           <argLine>-Xmx512m -Dtest.junit.threads=${surefire-forkcount} -Dzookeeper.junit.threadid=${surefire.forkNumber} -javaagent:${org.jmockit:jmockit:jar}</argLine>
           <basedir>${project.basedir}</basedir>
-          <redirectTestOutputToFile>true</redirectTestOutputToFile>
           <systemPropertyVariables>
             <build.test.dir>${project.build.directory}/surefire</build.test.dir>
             <zookeeper.DigestAuthenticationProvider.superDigest>super:D/InIHSb7yEEbrWz8b9l71RjZJU=</zookeeper.DigestAuthenticationProvider.superDigest>

zookeeper-server/src/main/java/org/apache/zookeeper/common/StringUtils.java

@@ -65,4 +65,13 @@ public class StringUtils {
         return list == null ? null : String.join(delim, list);
     }
 
+    /**
+     * Returns true if the string is null or it does not contain any non space characters.
+     * @param s the string
+     * @return true if the string is null or it does not contain any non space characters.
+     */
+    public static boolean isBlank(String s) {
+        return s == null || s.trim().isEmpty();
+    }
+
 }

zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxnFactory.java

@@ -603,6 +603,7 @@ public class NettyServerCnxnFactory extends ServerCnxnFactory {
         this.maxClientCnxns = maxClientCnxns;
         this.secure = secure;
         this.listenBacklog = backlog;
+        LOG.info("configure {} secure: {} on addr {}", this, secure, addr);
     }
 
     /** {@inheritDoc} */

zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServer.java

@@ -2172,4 +2172,9 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
     public boolean isReconfigEnabled() {
         return this.reconfigEnabled;
     }
+
+    public ZooKeeperServerShutdownHandler getZkShutdownHandler() {
+        return zkShutdownHandler;
+    }
+
 }

zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerMain.java

@@ -242,4 +242,26 @@ public class ZooKeeperServerMain {
         return secureCnxnFactory;
     }
 
+    /**
+     * Shutdowns properly the service, this method is not a public API.
+     */
+    public void close() {
+        ServerCnxnFactory primaryCnxnFactory = this.cnxnFactory;
+        if (primaryCnxnFactory == null) {
+            // in case of pure TLS we can hook into secureCnxnFactory
+            primaryCnxnFactory = secureCnxnFactory;
+        }
+        if (primaryCnxnFactory == null || primaryCnxnFactory.getZooKeeperServer() == null) {
+            return;
+        }
+        ZooKeeperServerShutdownHandler zkShutdownHandler = primaryCnxnFactory.getZooKeeperServer().getZkShutdownHandler();
+        zkShutdownHandler.handle(ZooKeeperServer.State.SHUTDOWN);
+        try {
+            // ServerCnxnFactory will call the shutdown
+            primaryCnxnFactory.join();
+        } catch (InterruptedException ex) {
+            Thread.currentThread().interrupt();
+        }
+    }
+
 }

zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerShutdownHandler.java

@@ -26,9 +26,9 @@ import org.apache.zookeeper.server.ZooKeeperServer.State;
  * SHUTDOWN server state transitions, which in turn releases the associated
  * shutdown latch.
  */
-class ZooKeeperServerShutdownHandler {
+public final class ZooKeeperServerShutdownHandler {
 
-    private final CountDownLatch shutdownLatch;
+        private final CountDownLatch shutdownLatch;
 
     ZooKeeperServerShutdownHandler(CountDownLatch shutdownLatch) {
         this.shutdownLatch = shutdownLatch;
@@ -39,7 +39,7 @@ class ZooKeeperServerShutdownHandler {
      *
      * @param state new server state
      */
-    void handle(State state) {
+    public void handle(State state) {
         if (state == State.ERROR || state == State.SHUTDOWN) {
             shutdownLatch.countDown();
         }

zookeeper-server/src/main/java/org/apache/zookeeper/server/embedded/ExitHandler.java

@@ -0,0 +1,35 @@
+package org.apache.zookeeper.server.embedded;
+
+/**
+ * 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.
+ */
+
+/**
+ * Behaviour of the server in case of internal error.
+ * When you are running tests you will use {@link #LOG_ONLY},
+ * but please take care of using {@link #EXIT} when runnning in production.
+ */
+public enum ExitHandler {
+    /**
+     * Exit the Java process.
+     */
+    EXIT,
+    /**
+     * Only log the error. This option is meant to be used only in tests.
+     */
+    LOG_ONLY;
+}

zookeeper-server/src/main/java/org/apache/zookeeper/server/embedded/ZooKeeperServerEmbedded.java

@@ -0,0 +1,118 @@
+package org.apache.zookeeper.server.embedded;
+
+/**
+ * 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.
+ */
+
+import java.nio.file.Path;
+import java.util.Objects;
+import java.util.Properties;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.apache.yetus.audience.InterfaceStability;
+
+/**
+ * This API allows you to start a ZooKeeper server node from Java code <p>
+ * The server will run inside the same process.<p>
+ * Typical usecases are:
+ * <ul>
+ * <li>Running automated tests</li>
+ * <li>Launch ZooKeeper server with a Java based service management system</li>
+ * </ul>
+ * <p>
+ * Please take into consideration that in production usually it is better to not run the client
+ * together with the server in order to avoid race conditions, especially around how ephemeral nodes work.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public interface ZooKeeperServerEmbedded extends AutoCloseable {
+    /**
+     * Builder for ZooKeeperServerEmbedded.
+     */
+    class ZookKeeperServerEmbeddedBuilder {
+
+        private Path baseDir;
+        private Properties configuration;
+        private ExitHandler exitHandler = ExitHandler.EXIT;
+
+        /**
+         * Base directory of the server.
+         * The system will create a temporary configuration file inside this directory.
+         * Please remember that dynamic configuration files wil be saved into this directory by default.
+         * <p>
+         * If you do not set a 'dataDir' configuration entry the system will use a subdirectory of baseDir.
+         * @param baseDir
+         * @return the builder
+         */
+        public ZookKeeperServerEmbeddedBuilder baseDir(Path baseDir) {
+            this.baseDir = Objects.requireNonNull(baseDir);
+            return this;
+        }
+
+        /**
+         * Set the contents of the main configuration as it would be in zk_server.conf file.
+         * @param configuration the configuration
+         * @return the builder
+         */
+        public ZookKeeperServerEmbeddedBuilder configuration(Properties configuration) {
+            this.configuration = Objects.requireNonNull(configuration);
+            return this;
+        }
+
+        /**
+         * Set the behaviour in case of hard system errors, see {@link ExitHandler}.
+         * @param exitHandler the handler
+         * @return the builder
+         */
+        public ZookKeeperServerEmbeddedBuilder exitHandler(ExitHandler exitHandler) {
+            this.exitHandler = Objects.requireNonNull(exitHandler);
+            return this;
+        }
+
+        /**
+         * Validate the configuration and create the server, without starting it.
+         * @return the new server
+         * @throws Exception
+         * @see #start()
+         */
+        public ZooKeeperServerEmbedded build() throws Exception {
+            if (baseDir == null) {
+                throw new IllegalStateException("baseDir is null");
+            }
+            if (configuration == null) {
+                throw new IllegalStateException("configuration is null");
+            }
+            return new ZooKeeperServerEmbeddedImpl(configuration, baseDir, exitHandler);
+        }
+    }
+
+    static ZookKeeperServerEmbeddedBuilder builder() {
+        return new ZookKeeperServerEmbeddedBuilder();
+    }
+
+    /**
+     * Start the server.
+     * @throws Exception
+     */
+    void start() throws Exception;
+
+    /**
+     * Shutdown gracefully the server and wait for resources to be released.
+     */
+    @Override
+    void close();
+
+}

zookeeper-server/src/main/java/org/apache/zookeeper/server/embedded/ZooKeeperServerEmbeddedImpl.java

@@ -0,0 +1,162 @@
+package org.apache.zookeeper.server.embedded;
+
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.zookeeper.server.DatadirCleanupManager;
+import org.apache.zookeeper.server.ExitCode;
+import org.apache.zookeeper.server.ServerConfig;
+import org.apache.zookeeper.server.ZooKeeperServerMain;
+import org.apache.zookeeper.server.quorum.QuorumPeer;
+import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
+import org.apache.zookeeper.server.quorum.QuorumPeerMain;
+import org.apache.zookeeper.util.ServiceUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * 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.
+ */
+/**
+ * Implementation of ZooKeeperServerEmbedded.
+ */
+class ZooKeeperServerEmbeddedImpl implements ZooKeeperServerEmbedded {
+
+    private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperServerEmbeddedImpl.class);
+
+    private final QuorumPeerConfig config;
+    private QuorumPeerMain maincluster;
+    private ZooKeeperServerMain mainsingle;
+    private Thread thread;
+    private DatadirCleanupManager purgeMgr;
+    private final ExitHandler exitHandler;
+    private volatile boolean stopping;
+
+    ZooKeeperServerEmbeddedImpl(Properties p, Path baseDir, ExitHandler exitHandler) throws Exception {
+        if (!p.containsKey("dataDir")) {
+            p.put("dataDir", baseDir.resolve("data").toAbsolutePath().toString());
+        }
+        Path configFile = Files.createTempFile(baseDir, "zookeeper.configuration", ".properties");
+        try (OutputStream oo = Files.newOutputStream(configFile)) {
+            p.store(oo, "Automatically generated at every-boot");
+        }
+        this.exitHandler = exitHandler;
+        LOG.info("Current configuration is at {}", configFile.toAbsolutePath());
+        config = new QuorumPeerConfig();
+        config.parse(configFile.toAbsolutePath().toString());
+        LOG.info("ServerID:" + config.getServerId());
+        LOG.info("DataDir:" + config.getDataDir());
+        LOG.info("Servers:" + config.getServers());
+        LOG.info("ElectionPort:" + config.getElectionPort());
+        LOG.info("SyncLimit:" + config.getSyncLimit());
+        LOG.info("PeerType:" + config.getPeerType());
+        LOG.info("Distributed:" + config.isDistributed());
+        LOG.info("SyncEnabled:" + config.getSyncEnabled());
+        LOG.info("MetricsProviderClassName:" + config.getMetricsProviderClassName());
+
+        for (Map.Entry<Long, QuorumPeer.QuorumServer> server : config.getServers().entrySet()) {
+            LOG.info("Server: " + server.getKey() + " -> addr " + server.getValue().addr + " elect "
+                    + server.getValue().electionAddr + " id=" + server.getValue().id + " type "
+                    + server.getValue().type);
+        }
+    }
+
+    @Override
+    public void start() throws Exception {
+        switch (exitHandler) {
+            case EXIT:
+                ServiceUtils.setSystemExitProcedure(ServiceUtils.SYSTEM_EXIT);
+                break;
+            case LOG_ONLY:
+                ServiceUtils.setSystemExitProcedure(ServiceUtils.LOG_ONLY);
+                break;
+            default:
+                ServiceUtils.setSystemExitProcedure(ServiceUtils.SYSTEM_EXIT);
+                break;
+        }
+
+
+        if (config.getServers().size() > 1 || config.isDistributed()) {
+            LOG.info("Running ZK Server in single Quorum MODE");
+
+            maincluster = new QuorumPeerMain();
+
+            // Start and schedule the the purge task
+            purgeMgr = new DatadirCleanupManager(config
+                    .getDataDir(), config.getDataLogDir(), config
+                    .getSnapRetainCount(), config.getPurgeInterval());
+            purgeMgr.start();
+
+            thread = new Thread("zkservermainrunner") {
+                @Override
+                public void run() {
+                    try {
+                        maincluster.runFromConfig(config);
+                        maincluster.close();
+                        LOG.info("ZK server died. Requsting stop on JVM");
+                        if (!stopping) {
+                            ServiceUtils.requestSystemExit(ExitCode.EXECUTION_FINISHED.getValue());
+                        }
+                    } catch (Throwable t) {
+                        LOG.error("error during server lifecycle", t);
+                        maincluster.close();
+                        if (!stopping) {
+                            ServiceUtils.requestSystemExit(ExitCode.INVALID_INVOCATION.getValue());
+                        }
+                    }
+                }
+            };
+            thread.start();
+        } else {
+            LOG.info("Running ZK Server in single STANDALONE MODE");
+            mainsingle = new ZooKeeperServerMain();
+            purgeMgr = new DatadirCleanupManager(config
+                    .getDataDir(), config.getDataLogDir(), config
+                    .getSnapRetainCount(), config.getPurgeInterval());
+            purgeMgr.start();
+            thread = new Thread("zkservermainrunner") {
+                @Override
+                public void run() {
+                    try {
+                        ServerConfig cc = new ServerConfig();
+                        cc.readFrom(config);
+                        LOG.info("ZK server starting");
+                        mainsingle.runFromConfig(cc);
+                        LOG.info("ZK server died. Requesting stop on JVM");
+                        if (!stopping) {
+                            ServiceUtils.requestSystemExit(ExitCode.EXECUTION_FINISHED.getValue());
+                        }
+                    } catch (Throwable t) {
+                        LOG.error("error during server lifecycle", t);
+                        mainsingle.close();
+                        if (!stopping) {
+                            ServiceUtils.requestSystemExit(ExitCode.INVALID_INVOCATION.getValue());
+                        }
+                    }
+                }
+            };
+            thread.start();
+        }
+    }
+
+    @Override
+    public void close() {
+        LOG.info("Stopping ZK Server");
+        stopping = true;
+        if (mainsingle != null) {
+            mainsingle.close();
+        }
+        if (maincluster != null) {
+            maincluster.close();
+        }
+    }
+}

zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/FileTxnSnapLog.java

@@ -611,14 +611,16 @@ public class FileTxnSnapLog {
      * @throws IOException
      */
     public void close() throws IOException {
-        if (txnLog != null) {
-            txnLog.close();
-            txnLog = null;
+        TxnLog txnLogToClose = txnLog;
+        if (txnLogToClose != null) {
+            txnLogToClose.close();
         }
-        if (snapLog != null) {
-            snapLog.close();
-            snapLog = null;
+        txnLog = null;
+        SnapShot snapSlogToClose = snapLog;
+        if (snapSlogToClose != null) {
+            snapSlogToClose.close();
         }
+        snapLog = null;
     }
 
     @SuppressWarnings("serial")

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumPeerMain.java

@@ -248,4 +248,23 @@ public class QuorumPeerMain {
         return new QuorumPeer();
     }
 
+    /**
+     * Shutdowns properly the service, this method is not a public API.
+     */
+    public void close() {
+        if (quorumPeer != null) {
+            try {
+                quorumPeer.shutdown();
+            } finally {
+                quorumPeer = null;
+            }
+        }
+    }
+
+    @Override
+    public String toString() {
+        QuorumPeer peer = quorumPeer;
+        return peer == null ? "" : peer.toString();
+    }
+
 }

zookeeper-server/src/main/java/org/apache/zookeeper/util/ServiceUtils.java

@@ -51,7 +51,7 @@ public abstract class ServiceUtils {
                 + "Actually System.exit is disabled", code);
     };
 
-    private static Consumer<Integer> systemExitProcedure = SYSTEM_EXIT;
+    private static volatile Consumer<Integer> systemExitProcedure = SYSTEM_EXIT;
 
     /**
      * Override system callback. Useful for preventing the JVM to exit in tests

zookeeper-server/src/test/java/org/apache/zookeeper/server/embedded/ZookeeperServeInfo.java

@@ -0,0 +1,300 @@
+/**
+ * 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.server.embedded;
+
+import java.lang.management.ManagementFactory;
+import java.lang.reflect.UndeclaredThrowableException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import javax.management.InstanceNotFoundException;
+import javax.management.MBeanServer;
+import javax.management.MBeanServerInvocationHandler;
+import javax.management.ObjectInstance;
+import javax.management.ObjectName;
+import org.apache.zookeeper.common.StringUtils;
+import org.apache.zookeeper.server.ConnectionMXBean;
+import org.apache.zookeeper.server.ZooKeeperServerBean;
+import org.apache.zookeeper.server.quorum.LocalPeerMXBean;
+import org.apache.zookeeper.server.quorum.QuorumBean;
+import org.apache.zookeeper.server.quorum.QuorumMXBean;
+import org.apache.zookeeper.server.quorum.RemotePeerMXBean;
+
+public final class ZookeeperServeInfo {
+
+    private static final MBeanServer localServer = ManagementFactory.getPlatformMBeanServer();
+
+    private ZookeeperServeInfo() {
+    }
+
+    public static class PeerInfo {
+
+        private final String name;
+        private final String quorumAddress;
+        private final String state;
+        private final boolean leader;
+
+        public PeerInfo(String name, String quorumAddress, String state, boolean leader) {
+            this.name = name;
+            this.quorumAddress = quorumAddress;
+            this.state = state;
+            this.leader = leader;
+        }
+
+        public String getName() {
+            return name;
+        }
+
+        public String getQuorumAddress() {
+            return quorumAddress;
+        }
+
+        public String getState() {
+            return state;
+        }
+
+        public boolean isLeader() {
+            return leader;
+        }
+
+        @Override
+        public String toString() {
+            return "PeerInfo{" + "name=" + name + ", leader=" + leader + ", quorumAddress=" + quorumAddress
+                    + ", state=" + state + '}';
+        }
+    }
+
+    public static class ConnectionInfo {
+
+        private final String sourceip;
+        private final String sessionid;
+        private final String lastoperation;
+        private final String lastResponseTime;
+        private final String avgLatency;
+        private final String lastLatency;
+        private final String nodes;
+
+        public ConnectionInfo(String sourceip, String sessionid, String lastoperation, String lastResponseTime,
+                              String avgLatency, String lastLatency, String nodes) {
+            this.sourceip = sourceip;
+            this.sessionid = sessionid;
+            this.lastoperation = lastoperation;
+            this.lastResponseTime = lastResponseTime;
+            this.avgLatency = avgLatency;
+            this.lastLatency = lastLatency;
+            this.nodes = nodes;
+        }
+
+        public String getLastLatency() {
+            return lastLatency;
+        }
+
+        public String getSourceip() {
+            return sourceip;
+        }
+
+        public String getSessionid() {
+            return sessionid;
+        }
+
+        public String getLastoperation() {
+            return lastoperation;
+        }
+
+        public String getLastResponseTime() {
+            return lastResponseTime;
+        }
+
+        public String getAvgLatency() {
+            return avgLatency;
+        }
+
+        public String getNodes() {
+            return nodes;
+        }
+
+        @Override
+        public String toString() {
+            return "ConnectionInfo{" + "sourceip=" + sourceip + ", sessionid=" + sessionid + ", lastoperation="
+                    + lastoperation + ", lastResponseTime=" + lastResponseTime + ", avgLatency=" + avgLatency
+                    + ", nodes=" + nodes + '}';
+        }
+    }
+
+    public static class ServerInfo {
+
+        private final List<ConnectionInfo> connections = new ArrayList<>();
+        private boolean leader;
+        private boolean standaloneMode;
+        public List<PeerInfo> peers = new ArrayList<>();
+
+        public boolean isStandaloneMode() {
+            return standaloneMode;
+        }
+
+        public List<ConnectionInfo> getConnections() {
+            return connections;
+        }
+
+        public boolean isLeader() {
+            return leader;
+        }
+
+        public List<PeerInfo> getPeers() {
+            return Collections.unmodifiableList(peers);
+        }
+
+        public void addPeer(PeerInfo peer) {
+            peers.add(peer);
+        }
+
+        @Override
+        public String toString() {
+            return "ServerInfo{" + "connections=" + connections + ", leader=" + leader + ", standaloneMode="
+                    + standaloneMode + ", peers=" + peers + '}';
+        }
+
+    }
+
+    public static ServerInfo getStatus() throws Exception {
+        return getStatus("*");
+    }
+
+    public static ServerInfo getStatus(String beanName) throws Exception {
+
+        ServerInfo info = new ServerInfo();
+        boolean standalonemode = false;
+        // org.apache.ZooKeeperService:name0=ReplicatedServer_id1,name1=replica.1,name2=Follower,name3=Connections,
+        // name4=10.168.10.119,name5=0x13e83353764005a
+        // org.apache.ZooKeeperService:name0=ReplicatedServer_id2,name1=replica.2,name2=Leader
+        if (StringUtils.isBlank(beanName)) {
+            beanName = "*";
+        }
+        ObjectName objectName = new ObjectName("org.apache.ZooKeeperService:name0=" + beanName);
+        Set<ObjectInstance> first_level_beans = localServer.queryMBeans(objectName, null);
+        if (first_level_beans.isEmpty()) {
+            throw new IllegalStateException("No ZooKeeper server found in this JVM with name " + objectName);
+        }
+        String myName = "";
+        for (ObjectInstance o : first_level_beans) {
+            if (o.getClassName().equalsIgnoreCase(ZooKeeperServerBean.class.getName())) {
+                standalonemode = true;
+                info.leader = true;
+                info.addPeer(new PeerInfo("local", "local", "STANDALONE", true));
+            } else if (o.getClassName().equalsIgnoreCase(QuorumBean.class.getName())) {
+                standalonemode = false;
+                try {
+                    QuorumMXBean quorum = MBeanServerInvocationHandler.newProxyInstance(localServer, o.getObjectName(),
+                            QuorumMXBean.class, false);
+                    myName = quorum.getName();
+                } catch (UndeclaredThrowableException err) {
+                    if (err.getCause() instanceof javax.management.InstanceNotFoundException) {
+                        // maybe server not yet started or already stopped ?
+                    } else {
+                        throw err;
+                    }
+                }
+            }
+        }
+        info.standaloneMode = standalonemode;
+        if (standalonemode) {
+            Set<ObjectInstance> connectionsbeans = localServer.queryMBeans(new ObjectName(
+                    "org.apache.ZooKeeperService:name0=*,name1=Connections,name2=*,name3=*"), null);
+            for (ObjectInstance conbean : connectionsbeans) {
+                ConnectionMXBean cc = MBeanServerInvocationHandler.
+                        newProxyInstance(localServer, conbean.getObjectName(), ConnectionMXBean.class, false);
+                try {
+                    String nodes = "";
+                    if (cc.getEphemeralNodes() != null) {
+                        nodes = Arrays.asList(cc.getEphemeralNodes()) + "";
+                    }
+                    info.connections.add(new ConnectionInfo(cc.getSourceIP(), cc.getSessionId(), cc.getLastOperation(),
+                            cc.getLastResponseTime(), cc.getAvgLatency() + "", cc.getLastLatency() + "", nodes));
+                } catch (Exception ex) {
+                    if (ex instanceof InstanceNotFoundException && ex.getCause() instanceof InstanceNotFoundException) {
+                        // SKIP
+                    } else {
+                        throw ex;
+                    }
+                }
+            }
+        } else {
+            if (myName.isEmpty()) {
+                throw new IllegalStateException(
+                        "Cannot find local JMX name for current node, in quorum mode, scanned " + first_level_beans);
+            }
+            boolean leader = false;
+            Set<ObjectInstance> replicas = localServer.queryMBeans(new ObjectName(
+                    "org.apache.ZooKeeperService:name0=" + myName + ",name1=*"), null);
+            for (ObjectInstance o : replicas) {
+                if (o.getClassName().toLowerCase().contains("local")) {
+                    LocalPeerMXBean local = MBeanServerInvocationHandler.
+                            newProxyInstance(localServer, o.getObjectName(), LocalPeerMXBean.class, false);
+                    info.addPeer(new PeerInfo(local.getName(), local.getQuorumAddress(), local.getState() + "",
+                            local.isLeader()));
+
+                    ObjectName asfollowername = new ObjectName(o.getObjectName() + ",name2=Follower");
+                    ObjectName asleadername = new ObjectName(o.getObjectName() + ",name2=Leader");
+                    boolean isleader = localServer.isRegistered(asleadername);
+                    Set<ObjectInstance> connectionsbeans = null;
+                    if (isleader) {
+                        leader = true;
+                        ObjectName asleaderconnections = new ObjectName(
+                                asleadername + ",name3=Connections,name4=*,name5=*");
+                        connectionsbeans = localServer.queryMBeans(asleaderconnections, null);
+                    } else {
+                        leader = false;
+                        ObjectName asfollowernameconnections = new ObjectName(
+                                asfollowername + ",name3=Connections,name4=*,name5=*");
+                        connectionsbeans = localServer.queryMBeans(asfollowernameconnections, null);
+                    }
+
+                    for (ObjectInstance conbean : connectionsbeans) {
+                        ConnectionMXBean cc = MBeanServerInvocationHandler.newProxyInstance(localServer,
+                                conbean.getObjectName(), ConnectionMXBean.class, false);
+                        try {
+                            String nodes = "";
+                            if (cc.getEphemeralNodes() != null) {
+                                nodes = Arrays.asList(cc.getEphemeralNodes()) + "";
+                            }
+                            info.connections.add(new ConnectionInfo(cc.getSourceIP(), cc.getSessionId(), cc.
+                                    getLastOperation(), cc.getLastResponseTime(), cc.getAvgLatency() + "", cc.
+                                    getLastLatency() + "", nodes));
+                        } catch (Exception ex) {
+                            if (ex instanceof InstanceNotFoundException && ex.getCause() instanceof InstanceNotFoundException) {
+                                // SKIP
+                            } else {
+                                throw ex;
+                            }
+                        }
+                    }
+                } else {
+                    RemotePeerMXBean remote = MBeanServerInvocationHandler.newProxyInstance(localServer, o.
+                            getObjectName(), RemotePeerMXBean.class, false);
+                    info.addPeer(new PeerInfo(remote.getName(), remote.getQuorumAddress(),
+                            "REMOTE", remote.isLeader()));
+                }
+
+            }
+            info.leader = leader;
+        }
+        return info;
+    }
+}

zookeeper-server/src/test/java/org/apache/zookeeper/server/embedded/ZookeeperServerClusterMutualAuthTest.java

@@ -0,0 +1,141 @@
+/**
+ * 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.server.embedded;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Properties;
+import javax.security.auth.login.Configuration;
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.test.ClientBase;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Test Quorum Mutual Auth with ZooKeeperEmbedded.
+ */
+public class ZookeeperServerClusterMutualAuthTest {
+
+    @BeforeAll
+    public static void setUpEnvironment() {
+        System.setProperty("java.security.auth.login.config", new File("src/test/resources/embedded/test_jaas_server_auth.conf")
+                .getAbsolutePath());
+        Configuration.getConfiguration().refresh();
+        System.setProperty("zookeeper.admin.enableServer", "false");
+        System.setProperty("zookeeper.4lw.commands.whitelist", "*");
+    }
+
+    @AfterAll
+    public static void cleanUpEnvironment() throws InterruptedException, IOException {
+        System.clearProperty("zookeeper.admin.enableServer");
+        System.clearProperty("zookeeper.4lw.commands.whitelist");
+        System.clearProperty("java.security.auth.login.config");
+        Configuration.getConfiguration().refresh();
+    }
+
+    @TempDir
+    public Path baseDir;
+
+    @Test
+    public void testStart() throws Exception {
+        Path baseDir1 = baseDir.resolve("server1");
+        Path baseDir2 = baseDir.resolve("server2");
+        Path baseDir3 = baseDir.resolve("server3");
+
+        int clientport1 = PortAssignment.unique();
+        int clientport2 = PortAssignment.unique();
+        int clientport3 = PortAssignment.unique();
+
+        int port4 = PortAssignment.unique();
+        int port5 = PortAssignment.unique();
+        int port6 = PortAssignment.unique();
+
+        int port7 = PortAssignment.unique();
+        int port8 = PortAssignment.unique();
+        int port9 = PortAssignment.unique();
+
+        Properties config = new Properties();
+        config.put("host", "localhost");
+        config.put("ticktime", "10");
+        config.put("initLimit", "4000");
+        config.put("syncLimit", "5");
+
+        config.put("server.1", "localhost:" + port4 + ":" + port7);
+        config.put("server.2", "localhost:" + port5 + ":" + port8);
+        config.put("server.3", "localhost:" + port6 + ":" + port9);
+
+        config.put("quorum.auth.enableSasl", "true");
+        config.put("quorum.auth.learnerRequireSasl", "true");
+        config.put("quorum.auth.serverRequireSasl", "true");
+        config.put("quorum.auth.learner.loginContext", "QuorumLearner");
+        config.put("quorum.auth.server.loginContext", "QuorumServer");
+        config.put("quorum.auth.kerberos.servicePrincipal", "servicename/_HOST");
+        config.put("quorum.cnxn.threads.size", "20");
+
+        final Properties configZookeeper1 = new Properties();
+        configZookeeper1.putAll(config);
+        configZookeeper1.put("clientPort", clientport1 + "");
+
+        final Properties configZookeeper2 = new Properties();
+        configZookeeper2.putAll(config);
+        configZookeeper2.put("clientPort", clientport2 + "");
+
+        final Properties configZookeeper3 = new Properties();
+        configZookeeper3.putAll(config);
+        configZookeeper3.put("clientPort", clientport3 + "");
+
+        Files.createDirectories(baseDir1.resolve("data"));
+        Files.write(baseDir1.resolve("data").resolve("myid"), "1".getBytes("ASCII"));
+        Files.createDirectories(baseDir2.resolve("data"));
+        Files.write(baseDir2.resolve("data").resolve("myid"), "2".getBytes("ASCII"));
+        Files.createDirectories(baseDir3.resolve("data"));
+        Files.write(baseDir3.resolve("data").resolve("myid"), "3".getBytes("ASCII"));
+
+        try (ZooKeeperServerEmbedded zkServer1 = ZooKeeperServerEmbedded.builder().configuration(configZookeeper1).baseDir(baseDir1).exitHandler(ExitHandler.LOG_ONLY).build();
+                ZooKeeperServerEmbedded zkServer2 = ZooKeeperServerEmbedded.builder().configuration(configZookeeper2).baseDir(baseDir2).exitHandler(ExitHandler.LOG_ONLY).build();
+                ZooKeeperServerEmbedded zkServer3 = ZooKeeperServerEmbedded.builder().configuration(configZookeeper3).baseDir(baseDir3).exitHandler(ExitHandler.LOG_ONLY).build();) {
+            zkServer1.start();
+            zkServer2.start();
+            zkServer3.start();
+
+            assertTrue(ClientBase.waitForServerUp("localhost:" + clientport1, 60000));
+            assertTrue(ClientBase.waitForServerUp("localhost:" + clientport2, 60000));
+            assertTrue(ClientBase.waitForServerUp("localhost:" + clientport3, 60000));
+
+            for (int i = 0; i < 100; i++) {
+                ZookeeperServeInfo.ServerInfo status = ZookeeperServeInfo.getStatus("ReplicatedServer*");
+                System.out.println("status:" + status);
+                if (status.isLeader() && !status.isStandaloneMode() && status.getPeers().size() == 3) {
+                    break;
+                }
+                Thread.sleep(100);
+            }
+            ZookeeperServeInfo.ServerInfo status = ZookeeperServeInfo.getStatus("ReplicatedServer*");
+            assertTrue(status.isLeader());
+            assertTrue(!status.isStandaloneMode());
+            assertEquals(3, status.getPeers().size());
+        }
+    }
+
+}

zookeeper-server/src/test/java/org/apache/zookeeper/server/embedded/ZookeeperServerClusterTest.java

@@ -0,0 +1,123 @@
+/**
+ * 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.server.embedded;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Properties;
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.test.ClientBase;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+public class ZookeeperServerClusterTest {
+
+    @BeforeAll
+    public static void setUpEnvironment() {
+        System.setProperty("zookeeper.admin.enableServer", "false");
+        System.setProperty("zookeeper.4lw.commands.whitelist", "*");
+    }
+
+    @AfterAll
+    public static void cleanUpEnvironment() throws InterruptedException, IOException {
+        System.clearProperty("zookeeper.admin.enableServer");
+        System.clearProperty("zookeeper.4lw.commands.whitelist");
+    }
+
+    @TempDir
+    public Path baseDir;
+
+    @Test
+    public void testStart() throws Exception {
+        Path baseDir1 = baseDir.resolve("server1");
+        Path baseDir2 = baseDir.resolve("server2");
+        Path baseDir3 = baseDir.resolve("server3");
+
+        int clientport1 = PortAssignment.unique();
+        int clientport2 = PortAssignment.unique();
+        int clientport3 = PortAssignment.unique();
+
+        int port4 = PortAssignment.unique();
+        int port5 = PortAssignment.unique();
+        int port6 = PortAssignment.unique();
+
+        int port7 = PortAssignment.unique();
+        int port8 = PortAssignment.unique();
+        int port9 = PortAssignment.unique();
+
+        Properties config = new Properties();
+        config.put("host", "localhost");
+        config.put("ticktime", "10");
+        config.put("initLimit", "4000");
+        config.put("syncLimit", "5");
+        config.put("server.1", "localhost:" + port4 + ":" + port7);
+        config.put("server.2", "localhost:" + port5 + ":" + port8);
+        config.put("server.3", "localhost:" + port6 + ":" + port9);
+
+
+        final Properties configZookeeper1 = new Properties();
+        configZookeeper1.putAll(config);
+        configZookeeper1.put("clientPort", clientport1 + "");
+
+        final Properties configZookeeper2 = new Properties();
+        configZookeeper2.putAll(config);
+        configZookeeper2.put("clientPort", clientport2 + "");
+
+        final Properties configZookeeper3 =  new Properties();
+        configZookeeper3.putAll(config);
+        configZookeeper3.put("clientPort", clientport3 + "");
+
+        Files.createDirectories(baseDir1.resolve("data"));
+        Files.write(baseDir1.resolve("data").resolve("myid"), "1".getBytes("ASCII"));
+        Files.createDirectories(baseDir2.resolve("data"));
+        Files.write(baseDir2.resolve("data").resolve("myid"), "2".getBytes("ASCII"));
+        Files.createDirectories(baseDir3.resolve("data"));
+        Files.write(baseDir3.resolve("data").resolve("myid"), "3".getBytes("ASCII"));
+
+        try (ZooKeeperServerEmbedded zkServer1 = ZooKeeperServerEmbedded.builder().configuration(configZookeeper1).baseDir(baseDir1).exitHandler(ExitHandler.LOG_ONLY).build();
+                ZooKeeperServerEmbedded zkServer2 = ZooKeeperServerEmbedded.builder().configuration(configZookeeper2).baseDir(baseDir2).exitHandler(ExitHandler.LOG_ONLY).build();
+                ZooKeeperServerEmbedded zkServer3 = ZooKeeperServerEmbedded.builder().configuration(configZookeeper3).baseDir(baseDir3).exitHandler(ExitHandler.LOG_ONLY).build();) {
+            zkServer1.start();
+            zkServer2.start();
+            zkServer3.start();
+
+            assertTrue(ClientBase.waitForServerUp("localhost:" + clientport1, 60000));
+            assertTrue(ClientBase.waitForServerUp("localhost:" + clientport2, 60000));
+            assertTrue(ClientBase.waitForServerUp("localhost:" + clientport3, 60000));
+            for (int i = 0; i < 100; i++) {
+                ZookeeperServeInfo.ServerInfo status = ZookeeperServeInfo.getStatus("ReplicatedServer*");
+                System.out.println("status:" + status);
+                if (status.isLeader() && !status.isStandaloneMode() && status.getPeers().size() == 3) {
+                    break;
+                }
+                Thread.sleep(100);
+            }
+            ZookeeperServeInfo.ServerInfo status = ZookeeperServeInfo.getStatus("ReplicatedServer*");
+            assertTrue(status.isLeader());
+            assertTrue(!status.isStandaloneMode());
+            assertEquals(3, status.getPeers().size());
+
+        }
+    }
+
+}

zookeeper-server/src/test/java/org/apache/zookeeper/server/embedded/ZookeeperServerEmbeddedTest.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.server.embedded;
+
+import static org.junit.Assert.assertTrue;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Properties;
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.test.ClientBase;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+public class ZookeeperServerEmbeddedTest {
+
+    @BeforeAll
+    public static void setUpEnvironment() {
+        System.setProperty("zookeeper.admin.enableServer", "false");
+        System.setProperty("zookeeper.4lw.commands.whitelist", "*");
+    }
+
+    @AfterAll
+    public static void cleanUpEnvironment() throws InterruptedException, IOException {
+        System.clearProperty("zookeeper.admin.enableServer");
+        System.clearProperty("zookeeper.4lw.commands.whitelist");
+    }
+
+    @TempDir
+    public Path baseDir;
+
+    @Test
+    public void testStart() throws Exception {
+        int clientPort = PortAssignment.unique();
+        final Properties configZookeeper = new Properties();
+        configZookeeper.put("clientPort", clientPort + "");
+        configZookeeper.put("host", "localhost");
+        configZookeeper.put("ticktime", "4000");
+        try (ZooKeeperServerEmbedded zkServer = ZooKeeperServerEmbedded
+                .builder()
+                .baseDir(baseDir)
+                .configuration(configZookeeper)
+                .exitHandler(ExitHandler.LOG_ONLY)
+                .build()) {
+            zkServer.start();
+            assertTrue(ClientBase.waitForServerUp("localhost:" + clientPort, 60000));
+            for (int i = 0; i < 100; i++) {
+                ZookeeperServeInfo.ServerInfo status = ZookeeperServeInfo.getStatus("StandaloneServer*");
+                if (status.isLeader() && status.isStandaloneMode()) {
+                    break;
+                }
+                Thread.sleep(100);
+            }
+            ZookeeperServeInfo.ServerInfo status = ZookeeperServeInfo.getStatus("StandaloneServer*");
+            assertTrue(status.isLeader());
+            assertTrue(status.isStandaloneMode());
+        }
+    }
+
+}

zookeeper-server/src/test/java/org/apache/zookeeper/server/embedded/ZookeeperServerSslEmbeddedTest.java

@@ -0,0 +1,121 @@
+/**
+ * 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.server.embedded;
+
+import static org.junit.Assert.assertTrue;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.apache.zookeeper.test.ClientBase;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+public class ZookeeperServerSslEmbeddedTest {
+
+    @BeforeAll
+    public static void setUpEnvironment() {
+        System.setProperty("zookeeper.admin.enableServer", "false");
+        System.setProperty("zookeeper.4lw.commands.whitelist", "*");
+    }
+
+    @AfterAll
+    public static void cleanUpEnvironment() throws InterruptedException, IOException {
+        System.clearProperty("zookeeper.admin.enableServer");
+        System.clearProperty("zookeeper.4lw.commands.whitelist");
+        System.clearProperty("zookeeper.ssl.trustStore.location");
+        System.clearProperty("zookeeper.ssl.trustStore.password");
+        System.clearProperty("zookeeper.ssl.trustStore.type");
+    }
+
+    @TempDir
+    public Path baseDir;
+
+    @Test
+    public void testStart() throws Exception {
+
+        int clientPort = PortAssignment.unique();
+        int clientSecurePort = PortAssignment.unique();
+
+        final Properties configZookeeper = new Properties();
+        configZookeeper.put("clientPort", clientPort + "");
+        configZookeeper.put("secureClientPort", clientSecurePort + "");
+        configZookeeper.put("host", "localhost");
+        configZookeeper.put("ticktime", "4000");
+        // Netty is required for TLS
+        configZookeeper.put("serverCnxnFactory", org.apache.zookeeper.server.NettyServerCnxnFactory.class.getName());
+
+        File testKeyStore = new File("src/test/resources/embedded/testKeyStore.jks");
+        File testTrustStore = new File("src/test/resources/embedded/testTrustStore.jks");
+        assertTrue(testKeyStore.isFile());
+        assertTrue(testTrustStore.isFile());
+        configZookeeper.put("ssl.keyStore.location", testKeyStore.getAbsolutePath());
+        configZookeeper.put("ssl.keyStore.password", "testpass");
+        configZookeeper.put("ssl.keyStore.type", "JKS");
+
+        System.setProperty("zookeeper.ssl.trustStore.location", testTrustStore.getAbsolutePath());
+        System.setProperty("zookeeper.ssl.trustStore.password", "testpass");
+        System.setProperty("zookeeper.ssl.trustStore.type", "JKS");
+
+        try (ZooKeeperServerEmbedded zkServer = ZooKeeperServerEmbedded
+                .builder()
+                .baseDir(baseDir)
+                .configuration(configZookeeper)
+                .exitHandler(ExitHandler.LOG_ONLY)
+                .build()) {
+            zkServer.start();
+            assertTrue(ClientBase.waitForServerUp("localhost:" + clientPort, 60000));
+            for (int i = 0; i < 100; i++) {
+                ZookeeperServeInfo.ServerInfo status = ZookeeperServeInfo.getStatus("StandaloneServer*");
+                if (status.isLeader() && status.isStandaloneMode()) {
+                    break;
+                }
+                Thread.sleep(100);
+            }
+            ZookeeperServeInfo.ServerInfo status = ZookeeperServeInfo.getStatus("StandaloneServer*");
+            assertTrue(status.isLeader());
+            assertTrue(status.isStandaloneMode());
+
+            CountDownLatch l = new CountDownLatch(1);
+            ZKClientConfig zKClientConfig = new ZKClientConfig();
+            zKClientConfig.setProperty("zookeeper.client.secure", "true");
+            // only netty supports TLS
+            zKClientConfig.setProperty("zookeeper.clientCnxnSocket", org.apache.zookeeper.ClientCnxnSocketNetty.class.getName());
+            try (ZooKeeper zk = new ZooKeeper("localhost:" + clientSecurePort, 60000, (WatchedEvent event) -> {
+                switch (event.getState()) {
+                    case SyncConnected:
+                        l.countDown();
+                        break;
+                }
+            }, zKClientConfig)) {
+                assertTrue(zk.getClientConfig().getBoolean(ZKClientConfig.SECURE_CLIENT));
+                assertTrue(l.await(10, TimeUnit.SECONDS));
+            }
+
+        }
+    }
+
+}

zookeeper-server/src/test/resources/embedded/test_jaas_server_auth.conf

@@ -0,0 +1,18 @@
+Server {
+       org.apache.zookeeper.server.auth.DigestLoginModule required
+       user_foo="bar";
+};
+Client {
+       org.apache.zookeeper.server.auth.DigestLoginModule required
+       username="foo"
+       password="bar";
+};
+QuorumServer {
+       org.apache.zookeeper.server.auth.DigestLoginModule required
+       user_test="test";
+};
+QuorumLearner {
+       org.apache.zookeeper.server.auth.DigestLoginModule required
+       username="test"
+       password="test";
+};