docs: @Override
      public void createAsync(final SocketDestination dest,
                              final KeyedResourcePool<SocketDestination, ClientRequestExecutor> pool)
              throws Exception {
          int numCreated = created.incrementAndGet();
          if(logger.isDebugEnabled())
              logger.debug("Creating socket " + numCreated + " for " + dest.getHost() + ":"
                           + dest.getPort() + " using protocol "
                           + dest.getRequestFormatType().getCode());
  
          SocketChannel socketChannel = null;
          ClientRequestExecutor clientRequestExecutor = null;
          long durationMs = 0;
  
          try {
              socketChannel = SocketChannel.open();
              socketChannel.socket().setReceiveBufferSize(this.socketBufferSize);
              socketChannel.socket().setSendBufferSize(this.socketBufferSize);
              socketChannel.socket().setTcpNoDelay(true);
              socketChannel.socket().setSoTimeout(soTimeoutMs);
              socketChannel.socket().setKeepAlive(this.socketKeepAlive);
              socketChannel.configureBlocking(false);
              socketChannel.connect(new InetSocketAddress(dest.getHost(), dest.getPort()));
  
              if(logger.isDebugEnabled()) {
                  logger.debug("Created socket " + numCreated + " for " + dest.getHost() + ":"
                               + dest.getPort() + " using protocol "
                               + dest.getRequestFormatType().getCode() + " after " + durationMs
                               + " ms.");
              }
  
              ClientRequestSelectorManager selectorManager = selectorManagers[counter.getAndIncrement()
                                                                              % selectorManagers.length];
  
              Selector selector = selectorManager.getSelector();
              clientRequestExecutor = new ClientRequestExecutor(selector,
                                                                socketChannel,
                                                                socketBufferSize,
                                                                idleConnectionTimeoutNs,
                                                                dest);
              int timeoutMs = this.getTimeout();
  
              ProtocolNegotiatorClientRequest protocolRequest = new ProtocolNegotiatorClientRequest(dest.getRequestFormatType());
  
              NonblockingStoreCallback callback = new NonblockingStoreCallback() {
  
                  @Override
                  public void requestComplete(Object result, long requestTime) {
                      if(result instanceof Exception) {
                          Exception e = (Exception) result;
                          /*
                           * There are 2 places where we can get a store timeout
                           * Exception
                           * 
                           * 1) While doing connect - the machine was up once, but
                           * not anymore. In that case, TCP SYN will be sent by
                           * the client, but server would not sent TCP ACK as it
                           * is dead.
                           * 
                           * 2) After connect doing Protocol Negotiation - Most
                           * likely the server and kernel is up, but the process
                           * is in a zombie state because of hard drive failure or
                           * stuck in shutdown or doing a GC. This can be
                           * intermittent or hard failures. Before this code
                           * change, if the process entered this state, Voldemort
                           * clients may not detect the failure immediately. They
                           * are treated as normal errors, instead of catastrophic
                           * erros.This was the reason before it is better to kill
                           * the process on a machine and let the machine stay up.
                           * After this code change they will be treated as
                           * connection failures ( catastrophic errors) to help
                           * recover the clients faster.
                           * 
                           * The second case can increase the false positives, but
                           * if a server is consistently timing out it is better
                           * to treat the server as dead and let the clients
                           * recover faster.
                           */
                          if(e instanceof StoreTimeoutException) {
                              e = new UnreachableStoreException("Error establishing connection for destination "
                                                                        + dest,
                                                                new ConnectException(e.getMessage()));
                          }
  
                          if(logger.isDebugEnabled()) {
                            logger.debug("Reporting exception to pool " + e.getClass()
                                        + " for destination " + dest);
                          }
  
                          pool.reportException(dest, e);
                      }
                  }
  
              };
  
              NonblockingStoreCallbackClientRequest<String> clientRequest = new NonblockingStoreCallbackClientRequest<String>(pool,
                                                                                                                              dest,
                                                                                                                              protocolRequest,
                                                                                                                              clientRequestExecutor,
                                                                                                                              callback,
                                                                                                                              stats);
  
              clientRequestExecutor.setConnectRequest(clientRequest, timeoutMs);
  
              selectorManager.add(clientRequestExecutor);
              selector.wakeup();
          } catch(Exception e) {
              // Make sure not to leak socketChannels
              if(socketChannel != null) {
                  try {
                      socketChannel.close();
                  } catch(Exception ex) {
                      if(logger.isEnabledFor(Level.WARN))
                          logger.warn(ex, ex);
                  }
              }
              // If clientRequestExector is not null, some additional clean up may
              // be warranted. However, clientRequestExecutor.close(), the
              // "obvious" clean up, is not safe to call here. This is because
              // .close() checks in a resource to the KeyedResourcePool that was
              // never checked out.
  
              throw UnreachableStoreException.wrap("Error establishing connection for destination "
                                                   + dest, e);
          }
          if(stats != null) {
              stats.incrementCount(dest, ClientSocketStats.Tracked.CONNECTION_CREATED_EVENT);
              stats.recordConnectionEstablishmentTimeUs(dest, durationMs * Time.US_PER_MS);
          }
      }