diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 4343093..4eb11fc 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -64,6 +64,9 @@
 * Deprecated `BytecodeUtil` and merged its functionality to the existing `BytecodeHelper`.
 * Added configuring implementation in HasStep
 * Remove static initialization for `GraphSONMessageSerializerV1d0` and `GraphSONMessageSerializerV1d0` in Java driver.
+* Connections to the server in a connection pool are created in parallel instead of serially in Java Driver.
+* Connection pools for multiple endpoints are created in parallel instead of serially in Java Driver.
+* Introduced new HostNotAvailable exception to represent cases when no server with active connection is available. 
 
 [[release-3-4-8]]
 === TinkerPop 3.4.8 (Release Date: August 3, 2020)
diff --git a/gremlin-driver/pom.xml b/gremlin-driver/pom.xml
index bd997a6..7bffa41 100644
--- a/gremlin-driver/pom.xml
+++ b/gremlin-driver/pom.xml
@@ -94,6 +94,16 @@
             <scope>test</scope>
         </dependency>
         <dependency>
+            <groupId>org.powermock</groupId>
+            <artifactId>powermock-module-junit4</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.powermock</groupId>
+            <artifactId>powermock-api-mockito</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
             <groupId>org.hamcrest</groupId>
             <artifactId>hamcrest-all</artifactId>
             <scope>test</scope>
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java
index 596910a..5460e75 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java
@@ -47,6 +47,7 @@
 import java.util.UUID;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import static java.lang.Math.toIntExact;
 
@@ -255,8 +256,16 @@
                 // forgot to enable it or perhaps the server is not configured for websockets.
                 handler.handshakeFuture().sync();
             } catch (Exception ex) {
-                throw new RuntimeException(new ConnectionException(connection.getUri(),
-                        "Could not complete websocket handshake - ensure that client protocol matches server", ex));
+                String errMsg = "";
+                if (ex instanceof TimeoutException) {
+                    errMsg = "Timed out while waiting to complete the connection setup. Consider increasing the " +
+                            "WebSocket handshake timeout duration.";
+                } else {
+                    errMsg = "Could not complete connection setup to the server. Ensure that SSL is correctly " +
+                            "configured at both the client and the server. Ensure that client WebSocket handshake " +
+                            "protocol matches the server. Ensure that the server is still reachable.";
+                }
+                throw new ConnectionException(connection.getUri(), errMsg, ex);
             }
         }
     }
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java
index 9cbc337..af4ab31 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java
@@ -18,7 +18,10 @@
  */
 package org.apache.tinkerpop.gremlin.driver;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.apache.tinkerpop.gremlin.driver.exception.ConnectionException;
+import org.apache.tinkerpop.gremlin.driver.exception.NoHostAvailableException;
 import org.apache.tinkerpop.gremlin.driver.message.RequestMessage;
 import org.apache.tinkerpop.gremlin.process.traversal.Bytecode;
 import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
@@ -26,8 +29,6 @@
 import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
 import org.apache.tinkerpop.gremlin.structure.Graph;
 import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.Collections;
 import java.util.HashMap;
@@ -37,11 +38,13 @@
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
 /**
@@ -105,7 +108,7 @@
      * one or more globally defined {@link Graph} or {@link TraversalSource} server bindings for the context of
      * the created {@code Client}.
      */
-    public Client alias(final Map<String,String> aliases) {
+    public Client alias(final Map<String, String> aliases) {
         return new AliasClusteredClient(this, aliases, settings);
     }
 
@@ -154,8 +157,7 @@
      * A version of {@link #submit(Bytecode)} which provides the ability to set per-request options.
      *
      * @param bytecode request in the form of gremlin {@link Bytecode}
-     * @param options for the request
-     *
+     * @param options  for the request
      * @see #submit(Bytecode)
      */
     public ResultSet submit(final Bytecode bytecode, final RequestOptions options) {
@@ -181,8 +183,7 @@
      * A version of {@link #submit(Bytecode)} which provides the ability to set per-request options.
      *
      * @param bytecode request in the form of gremlin {@link Bytecode}
-     * @param options for the request
-     *
+     * @param options  for the request
      * @see #submitAsync(Bytecode)
      */
     public CompletableFuture<ResultSet> submitAsync(final Bytecode bytecode, final RequestOptions options) {
@@ -201,8 +202,14 @@
         logger.debug("Initializing client on cluster [{}]", cluster);
 
         cluster.init();
+
         initializeImplementation();
 
+        // throw an error if no host is available even after initialization is complete.
+        if (cluster.availableHosts().isEmpty()) {
+            throw new NoHostAvailableException();
+        }
+
         initialized = true;
         return this;
     }
@@ -223,7 +230,7 @@
      * this method to concatenating a Gremlin script from dynamically produced strings and sending it to
      * {@link #submit(String)}.  Parameterized scripts will perform better.
      *
-     * @param gremlin the gremlin script to execute
+     * @param gremlin    the gremlin script to execute
      * @param parameters a map of parameters that will be bound to the script on execution
      */
     public ResultSet submit(final String gremlin, final Map<String, Object> parameters) {
@@ -263,7 +270,7 @@
      * The asynchronous version of {@link #submit(String, Map)}} where the returned future will complete when the
      * write of the request completes.
      *
-     * @param gremlin the gremlin script to execute
+     * @param gremlin    the gremlin script to execute
      * @param parameters a map of parameters that will be bound to the script on execution
      */
     public CompletableFuture<ResultSet> submitAsync(final String gremlin, final Map<String, Object> parameters) {
@@ -279,15 +286,15 @@
      * The asynchronous version of {@link #submit(String, Map)}} where the returned future will complete when the
      * write of the request completes.
      *
-     * @param gremlin the gremlin script to execute
-     * @param parameters a map of parameters that will be bound to the script on execution
+     * @param gremlin                the gremlin script to execute
+     * @param parameters             a map of parameters that will be bound to the script on execution
      * @param graphOrTraversalSource rebinds the specified global Gremlin Server variable to "g"
      * @deprecated As of release 3.4.0, replaced by {@link #submitAsync(String, RequestOptions)}.
      */
     @Deprecated
     public CompletableFuture<ResultSet> submitAsync(final String gremlin, final String graphOrTraversalSource,
                                                     final Map<String, Object> parameters) {
-        Map<String,String> aliases = null;
+        Map<String, String> aliases = null;
         if (graphOrTraversalSource != null && !graphOrTraversalSource.isEmpty()) {
             aliases = makeDefaultAliasMap(graphOrTraversalSource);
         }
@@ -299,15 +306,15 @@
      * The asynchronous version of {@link #submit(String, Map)}} where the returned future will complete when the
      * write of the request completes.
      *
-     * @param gremlin the gremlin script to execute
+     * @param gremlin    the gremlin script to execute
      * @param parameters a map of parameters that will be bound to the script on execution
-     * @param aliases aliases the specified global Gremlin Server variable some other name that then be used in the
-     *                script where the key is the alias name and the value represents the global variable on the
-     *                server
+     * @param aliases    aliases the specified global Gremlin Server variable some other name that then be used in the
+     *                   script where the key is the alias name and the value represents the global variable on the
+     *                   server
      * @deprecated As of release 3.4.0, replaced by {@link #submitAsync(String, RequestOptions)}.
      */
     @Deprecated
-    public CompletableFuture<ResultSet> submitAsync(final String gremlin, final Map<String,String> aliases,
+    public CompletableFuture<ResultSet> submitAsync(final String gremlin, final Map<String, String> aliases,
                                                     final Map<String, Object> parameters) {
         final RequestOptions.Builder options = RequestOptions.build();
         if (aliases != null && !aliases.isEmpty()) {
@@ -402,8 +409,8 @@
         return cluster;
     }
 
-    protected Map<String,String> makeDefaultAliasMap(final String graphOrTraversalSource) {
-        final Map<String,String> aliases = new HashMap<>();
+    protected Map<String, String> makeDefaultAliasMap(final String graphOrTraversalSource) {
+        final Map<String, String> aliases = new HashMap<>();
         aliases.put("g", graphOrTraversalSource);
         return aliases;
     }
@@ -443,8 +450,8 @@
          * this method to concatenating a Gremlin script from dynamically produced strings and sending it to
          * {@link #submit(String)}.  Parameterized scripts will perform better.
          *
-         * @param gremlin the gremlin script to execute
-         * @param parameters a map of parameters that will be bound to the script on execution
+         * @param gremlin                the gremlin script to execute
+         * @param parameters             a map of parameters that will be bound to the script on execution
          * @param graphOrTraversalSource rebinds the specified global Gremlin Server variable to "g"
          */
         public ResultSet submit(final String gremlin, final String graphOrTraversalSource, final Map<String, Object> parameters) {
@@ -460,7 +467,7 @@
          */
         @Override
         public Client alias(final String graphOrTraversalSource) {
-            final Map<String,String> aliases = new HashMap<>();
+            final Map<String, String> aliases = new HashMap<>();
             aliases.put("g", graphOrTraversalSource);
             return alias(aliases);
         }
@@ -469,7 +476,7 @@
          * {@inheritDoc}
          */
         @Override
-        public Client alias(final Map<String,String> aliases) {
+        public Client alias(final Map<String, String> aliases) {
             return new AliasClusteredClient(this, aliases, settings);
         }
 
@@ -492,7 +499,7 @@
             // you can get no possible hosts in more than a few situations. perhaps the servers are just all down.
             // or perhaps the client is not configured properly (disables ssl when ssl is enabled on the server).
             if (!possibleHosts.hasNext())
-                throw new TimeoutException("Timed out while waiting for an available host - check the client configuration and connectivity to the server if this message persists");
+                throw new NoHostAvailableException();
 
             final Host bestHost = possibleHosts.next();
             final ConnectionPool pool = hostConnectionPools.get(bestHost);
@@ -504,18 +511,20 @@
          */
         @Override
         protected void initializeImplementation() {
-            cluster.allHosts().forEach(host -> {
-                try {
-                    // hosts that don't initialize connection pools will come up as a dead host
-                    hostConnectionPools.put(host, new ConnectionPool(host, this));
-
-                    // added a new host to the cluster so let the load-balancer know
-                    this.cluster.loadBalancingStrategy().onNew(host);
-                } catch (Exception ex) {
-                    // catch connection errors and prevent them from failing the creation
-                    logger.warn("Could not initialize connection pool for {} - will try later", host);
+            try {
+                CompletableFuture.allOf(cluster.allHosts().stream()
+                        .map(host -> CompletableFuture.runAsync(() -> initializeConnectionSetupForHost.accept(host), cluster.executor()))
+                        .toArray(CompletableFuture[]::new))
+                        .join();
+            } catch (CompletionException ex) {
+                Throwable cause = null;
+                Throwable result = ex;
+                if (null != (cause = ex.getCause())) {
+                    result = cause;
                 }
-            });
+
+                logger.error("", result);
+            }
         }
 
         /**
@@ -526,11 +535,27 @@
             if (closing.get() != null)
                 return closing.get();
 
-            final CompletableFuture[] poolCloseFutures = new CompletableFuture[hostConnectionPools.size()];
-            hostConnectionPools.values().stream().map(ConnectionPool::closeAsync).collect(Collectors.toList()).toArray(poolCloseFutures);
-            closing.set(CompletableFuture.allOf(poolCloseFutures));
+            final CompletableFuture<Void> allPoolsClosedFuture =
+                    CompletableFuture.allOf(hostConnectionPools.values().stream()
+                            .map(ConnectionPool::closeAsync)
+                            .toArray(CompletableFuture[]::new));
+
+            closing.set(allPoolsClosedFuture);
             return closing.get();
         }
+
+        private Consumer<Host> initializeConnectionSetupForHost = host -> {
+            try {
+                // hosts that don't initialize connection pools will come up as a dead host
+                hostConnectionPools.put(host, new ConnectionPool(host, ClusteredClient.this));
+
+                // added a new host to the cluster so let the load-balancer know
+                ClusteredClient.this.cluster.loadBalancingStrategy().onNew(host);
+            } catch (RuntimeException ex) {
+                final String errMsg = "Could not initialize client for " + host;
+                throw new RuntimeException(errMsg, ex);
+            }
+        };
     }
 
     /**
@@ -539,10 +564,10 @@
      */
     public static class AliasClusteredClient extends Client {
         private final Client client;
-        private final Map<String,String> aliases = new HashMap<>();
+        private final Map<String, String> aliases = new HashMap<>();
         final CompletableFuture<Void> close = new CompletableFuture<>();
 
-        AliasClusteredClient(final Client client, final Map<String,String> aliases, final Client.Settings settings) {
+        AliasClusteredClient(final Client client, final Map<String, String> aliases, final Client.Settings settings) {
             super(client.cluster, settings);
             this.client = client;
             this.aliases.putAll(aliases);
@@ -559,8 +584,8 @@
                 // need to call buildMessage() right away to get client specific configurations, that way request specific
                 // ones can override as needed
                 final RequestMessage.Builder request = buildMessage(RequestMessage.build(Tokens.OPS_BYTECODE)
-                                                                                  .processor("traversal")
-                                                                                  .addArg(Tokens.ARGS_GREMLIN, bytecode));
+                        .processor("traversal")
+                        .addArg(Tokens.ARGS_GREMLIN, bytecode));
 
                 // apply settings if they were made available
                 options.getBatchSize().ifPresent(batchSize -> request.add(Tokens.ARGS_BATCH_SIZE, batchSize));
@@ -582,7 +607,7 @@
             // overrides which should be mucked with
             if (!aliases.isEmpty()) {
                 final Map original = (Map) msg.getArgs().getOrDefault(Tokens.ARGS_ALIASES, Collections.emptyMap());
-                aliases.forEach((k,v) -> {
+                aliases.forEach((k, v) -> {
                     if (!original.containsKey(k))
                         builder.addArg(Tokens.ARGS_ALIASES, aliases);
                 });
@@ -708,7 +733,12 @@
             if (hosts.isEmpty()) throw new IllegalStateException("No available host in the cluster");
             Collections.shuffle(hosts);
             final Host host = hosts.get(0);
-            connectionPool = new ConnectionPool(host, this, Optional.of(1), Optional.of(1));
+
+            try {
+                connectionPool = new ConnectionPool(host, this, Optional.of(1), Optional.of(1));
+            } catch (RuntimeException ex) {
+                logger.error("Could not initialize client for {}", host, ex);
+            }
         }
 
         @Override
@@ -758,7 +788,8 @@
         public static class Builder {
             private Optional<SessionSettings> session = Optional.empty();
 
-            private Builder() {}
+            private Builder() {
+            }
 
             /**
              * Enables a session. By default this will create a random session name and configure transactions to be
@@ -842,7 +873,8 @@
             private String sessionId = UUID.randomUUID().toString();
             private boolean forceClosed = false;
 
-            private Builder() {}
+            private Builder() {
+            }
 
             /**
              * If enabled, transactions will be "managed" such that each request will represent a complete transaction.
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
index 9636df9..26699bc 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
@@ -101,7 +101,7 @@
         this.maxInProcess = maxInProcess;
         this.keepAliveInterval = pool.settings().keepAliveInterval;
 
-        connectionLabel = String.format("Connection{host=%s}", pool.host);
+        connectionLabel = "Connection{host=" + pool.host + "}";
 
         if (cluster.isClosing())
             throw new IllegalStateException("Cannot open a connection with the cluster after close() is called");
@@ -128,9 +128,8 @@
             channel.closeFuture().addListener(new ChannelFutureListener() {
                 @Override
                 public void operationComplete(ChannelFuture future) throws Exception {
-                    if(logger.isDebugEnabled()) {
-                        logger.debug("OnChannelClose callback called for channel {}", channel.id().asShortText());
-                    }
+                    logger.debug("OnChannelClose callback called for channel {}", channel);
+
                     // Replace the channel if it was not intentionally closed using CloseAsync method.
                     if (thisConnection.closeFuture.get() == null) {
                         // delegate the task to worker thread and free up the event loop
@@ -143,11 +142,11 @@
 
             // Default WebSocketChannelizer uses Netty's IdleStateHandler
             if (!(channelizer instanceof Channelizer.WebSocketChannelizer)) {
+                logger.debug("Using custom keep alive handler.");
                 scheduleKeepAlive();
             }
-        } catch (Exception ie) {
-            logger.debug("Error opening connection on {}", uri);
-            throw new ConnectionException(uri, "Could not open connection", ie);
+        } catch (Exception ex) {
+            throw new ConnectionException(uri, "Could not open " + this.toString(), ex);
         }
     }
 
@@ -273,6 +272,7 @@
 
         // Default WebSocketChannelizer uses Netty's IdleStateHandler
         if (!(channelizer instanceof Channelizer.WebSocketChannelizer)) {
+            logger.debug("Using custom keep alive handler.");
             scheduleKeepAlive();
         }
 
@@ -409,10 +409,10 @@
     /**
      * Returns the short ID for the underlying channel for this connection.
      * <p>
-     * Currently only used for testing.
+     * Visible for testing.
      */
     String getChannelId() {
-        return (channel != null) ? channel.id().asShortText() : "";
+        return (channel != null) ? channel.id().asShortText() : "null";
     }
 
     @Override
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ConnectionPool.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ConnectionPool.java
index f7b35e9..295c9d0 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ConnectionPool.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ConnectionPool.java
@@ -18,25 +18,23 @@
  */
 package org.apache.tinkerpop.gremlin.driver;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.apache.tinkerpop.gremlin.driver.exception.ConnectionException;
 import org.apache.tinkerpop.gremlin.driver.message.RequestMessage;
 import org.apache.tinkerpop.gremlin.util.TimeUtil;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelFutureListener;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.Condition;
@@ -85,7 +83,7 @@
         this.host = host;
         this.client = client;
         this.cluster = client.cluster;
-        poolLabel = String.format("Connection Pool {host=%s}", host);
+        poolLabel = "Connection Pool {host=" + host + "}";
 
         final Settings.ConnectionPoolSettings settings = settings();
         this.minPoolSize = overrideMinPoolSize.orElse(settings.minSize);
@@ -97,15 +95,38 @@
         this.connections = new CopyOnWriteArrayList<>();
 
         try {
+            final List<CompletableFuture<Void>> connCreationFutures = new ArrayList<>();
             for (int i = 0; i < minPoolSize; i++) {
-                this.connections.add(new Connection(host.getHostUri(), this, settings.maxInProcessPerConnection));
+                connCreationFutures.add(CompletableFuture.runAsync(() -> {
+                    try {
+                        this.connections.add(new Connection(host.getHostUri(), this, settings.maxInProcessPerConnection));
+                    } catch (ConnectionException e) {
+                        throw new CompletionException(e);
+                    }
+                }, cluster.executor()));
             }
 
-        } catch (ConnectionException ce) {
-            // ok if we don't get it initialized here - when a request is attempted in a connection from the
-            // pool it will try to create new connections as needed.
-            logger.info("Could not initialize connections in pool for {} - pool size at {}", host, this.connections.size(), ce);
-            considerHostUnavailable();
+            CompletableFuture.allOf(connCreationFutures.toArray(new CompletableFuture[0])).join();
+        } catch (CancellationException ce) {
+            logger.warn("Initialization of connections cancelled for {}", getPoolInfo(), ce);
+            throw ce;
+        } catch (CompletionException ce) {
+            // Some connections might have been initialized. Close the connection pool gracefully to close them.
+            this.closeAsync();
+
+            final String errMsg = "Could not initialize " + minPoolSize + " (minPoolSize) connections in pool." +
+                    " Successful connections=" + this.connections.size() +
+                    ". Closing the connection pool.";
+
+
+            Throwable cause = null;
+            Throwable result = ce;
+
+            if (null != (cause = result.getCause())) {
+                result = cause;
+            }
+
+            throw new CompletionException(errMsg, result);
         }
 
         this.open = new AtomicInteger(connections.size());
@@ -316,7 +337,7 @@
         try {
             connections.add(new Connection(host.getHostUri(), this, settings().maxInProcessPerConnection));
         } catch (ConnectionException ce) {
-            logger.debug("Connections were under max, but there was an error creating the connection.", ce);
+            logger.error("Connections were under max, but there was an error creating the connection.", ce);
             open.decrementAndGet();
             considerHostUnavailable();
             return false;
@@ -350,7 +371,7 @@
 
         // only close the connection for good once it is done being borrowed or when it is dead
         if (connection.isDead() || connection.borrowed.get() == 0) {
-            if(bin.remove(connection)) {
+            if (bin.remove(connection)) {
                 connection.closeAsync();
                 // TODO: Log the following message on completion of the future returned by closeAsync.
                 logger.debug("{} destroyed", connection.getConnectionInfo());
@@ -396,7 +417,7 @@
             logger.debug("Continue to wait for connection on {} if {} > 0", host, remaining);
         } while (remaining > 0);
 
-        logger.debug("Timed-out waiting for connection on {} - possibly unavailable", host);
+        logger.error("Timed-out ({} {}) waiting for connection on {} - possibly unavailable", timeout, unit, host);
 
         // if we timeout borrowing a connection that might mean the host is dead (or the timeout was super short).
         // either way supply a function to reconnect
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/exception/ConnectionException.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/exception/ConnectionException.java
index 67101b3..551c9d0 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/exception/ConnectionException.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/exception/ConnectionException.java
@@ -23,9 +23,11 @@
 import java.util.Optional;
 
 /**
+ * This exception signifies network connection failure.
+ *
  * @author Stephen Mallette (http://stephen.genoprime.com)
  */
-public class ConnectionException extends Exception {
+public class ConnectionException extends RuntimeException {
     private URI uri;
     private InetSocketAddress address;
 
@@ -35,6 +37,12 @@
         this.uri = uri;
     }
 
+    public ConnectionException(final URI uri, final Throwable cause) {
+        super(cause);
+        this.uri = uri;
+        this.address = null;
+    }
+
     public ConnectionException(final URI uri, final String message, final Throwable cause) {
         super(message, cause);
         this.uri = uri;
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/exception/NoHostAvailableException.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/exception/NoHostAvailableException.java
new file mode 100644
index 0000000..e8b3b87
--- /dev/null
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/exception/NoHostAvailableException.java
@@ -0,0 +1,31 @@
+/*
+ * 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.tinkerpop.gremlin.driver.exception;
+
+public class NoHostAvailableException extends RuntimeException {
+
+    public NoHostAvailableException() {
+        super("All hosts are considered unavailable due to previous exceptions. Check the error log to find the actual reason.");
+    }
+
+    @Override
+    public synchronized Throwable fillInStackTrace() {
+        return this;
+    }
+}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketClientHandler.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketClientHandler.java
index 7e55b92..3d2df78 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketClientHandler.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketClientHandler.java
@@ -70,8 +70,10 @@
         if (!handshakeFuture.isDone()) {
             // channel was closed before the handshake could be completed.
             handshakeFuture.setFailure(
-                    new RuntimeException(String.format("Channel=[%s] closed before the handshake could complete",
-                            ctx.channel().toString())));
+                    new RuntimeException(String.format("WebSocket channel=[%s] closed before the handshake could complete." +
+                                    " Server logs could contain the reason for abrupt connection disconnect or the " +
+                                    "server might not be reachable from the client anymore.",
+                            ctx.channel().id().asShortText())));
         }
 
         super.channelInactive(ctx);
@@ -82,7 +84,7 @@
         if (event instanceof IdleStateEvent) {
             IdleStateEvent e = (IdleStateEvent) event;
             if (e.state() == IdleState.READER_IDLE) {
-                logger.warn("WebSocket connection " + ctx.channel() + " has been idle for too long.");
+                logger.warn("WebSocket connection {} has been idle for too long.", ctx.channel());
             } else if (e.state() == IdleState.WRITER_IDLE) {
                 logger.debug("Sending ping frame to the server");
                 ctx.writeAndFlush(new PingWebSocketFrame());
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientTest.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientTest.java
new file mode 100644
index 0000000..20f38ba
--- /dev/null
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.tinkerpop.gremlin.driver;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import org.apache.tinkerpop.gremlin.driver.exception.NoHostAvailableException;
+
+import java.util.Collections;
+import java.util.Optional;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
+import static org.mockito.Mockito.when;
+import static org.powermock.api.mockito.PowerMockito.whenNew;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({Client.ClusteredClient.class, Client.SessionedClient.class, Host.class, Cluster.class})
+public class ClientTest {
+    @Mock
+    private Cluster cluster;
+
+    @Mock
+    private Host mockAvailableHost;
+
+    @Mock
+    private Client.Settings settings;
+
+    private ScheduledExecutorService executor;
+
+    @Before
+    public void setup() {
+        executor = Executors.newScheduledThreadPool(1);
+        when(mockAvailableHost.isAvailable()).thenReturn(true);
+        when(cluster.allHosts()).thenReturn(Collections.singletonList(mockAvailableHost));
+        when(cluster.executor()).thenReturn(executor);
+    }
+
+    @After
+    public void cleanup() {
+        executor.shutdown();
+    }
+
+    @Test(expected = NoHostAvailableException.class)
+    public void shouldThrowErrorWhenConnPoolInitFailsForClusteredClient() throws Exception {
+        Client.ClusteredClient client = new Client.ClusteredClient(cluster, settings);
+        whenNew(ConnectionPool.class).withAnyArguments().thenThrow(new RuntimeException("cannot initialize client"));
+        client.init();
+    }
+
+    @Test(expected = NoHostAvailableException.class)
+    public void shouldThrowErrorWhenConnPoolInitFailsForSessionClient() throws Exception {
+        final Client.SessionSettings sessionSettings = Client.SessionSettings.build().sessionId("my-session-id").create();
+        when(settings.getSession()).thenReturn(Optional.of(sessionSettings));
+        Client.SessionedClient client = new Client.SessionedClient(cluster, settings);
+        whenNew(ConnectionPool.class).withAnyArguments().thenThrow(new RuntimeException("cannot initialize client"));
+        client.init();
+    }
+
+}
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ser/binary/types/sample/SamplePersonSerializerTest.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ser/binary/types/sample/SamplePersonSerializerTest.java
index 3d6bef3..1008653 100644
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ser/binary/types/sample/SamplePersonSerializerTest.java
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ser/binary/types/sample/SamplePersonSerializerTest.java
@@ -108,6 +108,7 @@
         final ResponseMessage deserialized = serializer.deserializeResponse(serialized);
 
         final SamplePerson actual = (SamplePerson) deserialized.getResult().getData();
+
         assertThat(actual, new ReflectionEquals(person));
     }
 
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/OpSelectorHandler.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/OpSelectorHandler.java
index 5a2e1f8..4259d15 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/OpSelectorHandler.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/OpSelectorHandler.java
@@ -102,10 +102,10 @@
             // periodically ping the server, but coming from this direction allows the server to kill channels that
             // have dead clients on the other end
             if (e.state() == IdleState.READER_IDLE) {
-                logger.info("Closing channel - client is disconnected after idle period of " + settings.idleConnectionTimeout + " " + ctx.channel());
+                logger.info("Closing channel - client is disconnected after idle period of " + settings.idleConnectionTimeout + " " + ctx.channel().id().asShortText());
                 ctx.close();
             } else if (e.state() == IdleState.WRITER_IDLE && settings.keepAliveInterval > 0) {
-                logger.info("Checking channel - sending ping to client after idle period of " + settings.keepAliveInterval + " " + ctx.channel());
+                logger.info("Checking channel - sending ping to client after idle period of " + settings.keepAliveInterval + " " + ctx.channel().id().asShortText());
                 ctx.writeAndFlush(channelizer.createIdleDetectionMessage());
             }
         }
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverIntegrateTest.java
index db88ef9..623107f 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverIntegrateTest.java
@@ -28,6 +28,7 @@
 import org.apache.tinkerpop.gremlin.driver.Result;
 import org.apache.tinkerpop.gremlin.driver.ResultSet;
 import org.apache.tinkerpop.gremlin.driver.Tokens;
+import org.apache.tinkerpop.gremlin.driver.exception.NoHostAvailableException;
 import org.apache.tinkerpop.gremlin.driver.exception.ResponseException;
 import org.apache.tinkerpop.gremlin.driver.handler.WebSocketClientHandler;
 import org.apache.tinkerpop.gremlin.driver.message.RequestMessage;
@@ -64,6 +65,7 @@
 
 import java.awt.Color;
 import java.io.File;
+import java.net.ConnectException;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -411,7 +413,7 @@
                 fail("Should not have gone through because the server is not running");
             } catch (Exception i) {
                 final Throwable root = ExceptionUtils.getRootCause(i);
-                assertThat(root, instanceOf(TimeoutException.class));
+                assertThat(root, instanceOf(NoHostAvailableException.class));
             }
 
             startServer();
@@ -447,7 +449,7 @@
             fail("Should not have gone through because the server is not running");
         } catch (Exception i) {
             final Throwable root = ExceptionUtils.getRootCause(i);
-            assertThat(root, instanceOf(TimeoutException.class));
+            assertThat(root, instanceOf(NoHostAvailableException.class));
         }
 
         startServer();
@@ -476,6 +478,7 @@
 
         try {
             final Client client = cluster.connect();
+            client.init();
 
             // the first host is dead on init.  request should succeed on localhost
             assertEquals(2, client.submit("1+1").all().join().get(0).getInt());
@@ -861,12 +864,29 @@
         final Cluster cluster = TestClientFactory.open();
         try {
             assertEquals(0, cluster.availableHosts().size());
-            cluster.connect().init();
+            final Client client1 = cluster.connect().init();
             assertEquals(1, cluster.availableHosts().size());
 
             stopServer();
 
+            // We create a new client here which will fail to initialize but the original client still has
+            // host marked as connected. Since the second client failed during initialization, it has no way to
+            // test if a host is indeed unreachable because it doesn't have any established connections. It will not add
+            // the host to load balancer but it will also not remove it if it already exists there. Leave that
+            // responsibility to a client that added it. In this case, let the second client perform it's own mechanism
+            // to mark host as unavailable. The first client will discover that the host has failed either with next
+            // keepAlive message or the next request, whichever is earlier. In this case, we will simulate the second
+            // scenario by sending a new request on first client. The request would fail (since server is down) and
+            // client should mark the host unavailable.
             cluster.connect().init();
+
+            try {
+                client1.submit("1+1").all().join();
+                fail("Expecting an exception because the server is shut down.");
+            } catch (Exception ex) {
+                // ignore the exception
+            }
+
             assertEquals(0, cluster.availableHosts().size());
         } finally {
             cluster.close();
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerAuthIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerAuthIntegrateTest.java
index 807e9a7..f1740c4 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerAuthIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerAuthIntegrateTest.java
@@ -21,6 +21,7 @@
 import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.apache.tinkerpop.gremlin.driver.Client;
 import org.apache.tinkerpop.gremlin.driver.Cluster;
+import org.apache.tinkerpop.gremlin.driver.exception.NoHostAvailableException;
 import org.apache.tinkerpop.gremlin.driver.exception.ResponseException;
 import org.apache.tinkerpop.gremlin.server.auth.SimpleAuthenticator;
 import org.ietf.jgss.GSSException;
@@ -88,8 +89,7 @@
             fail("This should not succeed as the client did not enable SSL");
         } catch(Exception ex) {
             final Throwable root = ExceptionUtils.getRootCause(ex);
-            assertEquals(TimeoutException.class, root.getClass());
-            assertThat(root.getMessage(), startsWith("Timed out while waiting for an available host"));
+            assertEquals(NoHostAvailableException.class, root.getClass());
         } finally {
             cluster.close();
         }
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerSslIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerSslIntegrateTest.java
index ab75b8e..8e1e8fb 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerSslIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerSslIntegrateTest.java
@@ -27,6 +27,7 @@
 import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.apache.tinkerpop.gremlin.driver.Client;
 import org.apache.tinkerpop.gremlin.driver.Cluster;
+import org.apache.tinkerpop.gremlin.driver.exception.NoHostAvailableException;
 import org.junit.Test;
 
 import java.util.Arrays;
@@ -203,7 +204,7 @@
             fail("Should throw exception because ssl is enabled on the server but not on client");
         } catch(Exception x) {
             final Throwable root = ExceptionUtils.getRootCause(x);
-            assertThat(root, instanceOf(TimeoutException.class));
+            assertThat(root, instanceOf(NoHostAvailableException.class));
         } finally {
             cluster.close();
         }
@@ -233,7 +234,7 @@
             fail("Should throw exception because ssl client auth is enabled on the server but client does not have a cert");
         } catch(Exception x) {
             final Throwable root = ExceptionUtils.getRootCause(x);
-            assertThat(root, instanceOf(TimeoutException.class));
+            assertThat(root, instanceOf(NoHostAvailableException.class));
         } finally {
             cluster.close();
         }
@@ -251,7 +252,7 @@
             fail("Should throw exception because ssl client auth is enabled on the server but does not trust client's cert");
         } catch(Exception x) {
             final Throwable root = ExceptionUtils.getRootCause(x);
-            assertThat(root, instanceOf(TimeoutException.class));
+            assertThat(root, instanceOf(NoHostAvailableException.class));
         } finally {
             cluster.close();
         }
@@ -294,7 +295,7 @@
             fail("Should throw exception because ssl client auth is enabled on the server but client does not have a cert");
         } catch (Exception x) {
             final Throwable root = ExceptionUtils.getRootCause(x);
-            assertThat(root, instanceOf(TimeoutException.class));
+            assertThat(root, instanceOf(NoHostAvailableException.class));
         } finally {
             cluster.close();
         }
@@ -311,7 +312,7 @@
             fail("Should throw exception because ssl client auth is enabled on the server but does not trust client's cert");
         } catch (Exception x) {
             final Throwable root = ExceptionUtils.getRootCause(x);
-            assertThat(root, instanceOf(TimeoutException.class));
+            assertThat(root, instanceOf(NoHostAvailableException.class));
         } finally {
             cluster.close();
         }
@@ -328,7 +329,7 @@
             fail("Should throw exception because ssl client requires TLSv1.2 whereas server supports only TLSv1.1");
         } catch (Exception x) {
             final Throwable root = ExceptionUtils.getRootCause(x);
-            assertThat(root, instanceOf(TimeoutException.class));
+            assertThat(root, instanceOf(NoHostAvailableException.class));
         } finally {
             cluster.close();
         }
@@ -345,7 +346,7 @@
             fail("Should throw exception because ssl client requires TLSv1.2 whereas server supports only TLSv1.1");
         } catch (Exception x) {
             final Throwable root = ExceptionUtils.getRootCause(x);
-            assertThat(root, instanceOf(TimeoutException.class));
+            assertThat(root, instanceOf(NoHostAvailableException.class));
         } finally {
             cluster.close();
         }
@@ -391,7 +392,7 @@
             fail("Should throw exception because incorrect keyStoreType is specified");
         } catch (Exception x) {
             final Throwable root = ExceptionUtils.getRootCause(x);
-            assertThat(root, instanceOf(TimeoutException.class));
+            assertThat(root, instanceOf(NoHostAvailableException.class));
         } finally {
             cluster.close();
         }
@@ -410,7 +411,7 @@
             fail("Should throw exception because incorrect trustStoreType is specified");
         } catch (Exception x) {
             final Throwable root = ExceptionUtils.getRootCause(x);
-            assertThat(root, instanceOf(TimeoutException.class));
+            assertThat(root, instanceOf(NoHostAvailableException.class));
         } finally {
             cluster.close();
         }
diff --git a/pom.xml b/pom.xml
index 3a5451b..8fd981c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -160,6 +160,7 @@
         <slf4j.version>1.7.25</slf4j.version>
         <snakeyaml.version>1.27</snakeyaml.version>
         <spark.version>2.4.0</spark.version>
+        <powermock.version>1.6.4</powermock.version>
 
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
@@ -721,6 +722,18 @@
                 </exclusions>
             </dependency>
             <dependency>
+                <groupId>org.powermock</groupId>
+                <artifactId>powermock-module-junit4</artifactId>
+                <version>${powermock.version}</version>
+                <scope>test</scope>
+            </dependency>
+            <dependency>
+                <groupId>org.powermock</groupId>
+                <artifactId>powermock-api-mockito</artifactId>
+                <version>${powermock.version}</version>
+                <scope>test</scope>
+            </dependency>
+            <dependency>
                 <groupId>org.hamcrest</groupId>
                 <artifactId>hamcrest-all</artifactId>
                 <version>1.3</version>