docs: @Nullable @GuardedBy("lock")
      protected Peer connectTo(PeerAddress address, boolean incrementMaxConnections, int connectTimeoutMillis) {
          checkState(lock.isHeldByCurrentThread());
          VersionMessage ver = getVersionMessage().duplicate();
          ver.bestHeight = chain == null ? 0 : chain.getBestChainHeight();
          ver.time = Utils.currentTimeSeconds();
          ver.receivingAddr = address;
          ver.receivingAddr.setParent(ver);
  
          Peer peer = createPeer(address, ver);
          peer.addConnectedEventListener(Threading.SAME_THREAD, startupListener);
          peer.addDisconnectedEventListener(Threading.SAME_THREAD, startupListener);
          peer.setMinProtocolVersion(vMinRequiredProtocolVersion);
          pendingPeers.add(peer);
  
          try {
              log.info("Attempting connection to {}     ({} connected, {} pending, {} max)", address,
                      peers.size(), pendingPeers.size(), maxConnections);
              ListenableFuture<SocketAddress> future = channels.openConnection(address.toSocketAddress(), peer);
              if (future.isDone())
                  Uninterruptibles.getUninterruptibly(future);
          } catch (ExecutionException e) {
              Throwable cause = Throwables.getRootCause(e);
              log.warn("Failed to connect to " + address + ": " + cause.getMessage());
              handlePeerDeath(peer, cause);
              return null;
          }
          peer.setSocketTimeout(connectTimeoutMillis);
          // When the channel has connected and version negotiated successfully, handleNewPeer will end up being called on
          // a worker thread.
          if (incrementMaxConnections) {
              // We don't use setMaxConnections here as that would trigger a recursive attempt to establish a new
              // outbound connection.
              maxConnections++;
          }
          return peer;
      }