private void connectToDS() {
    if (!isListening()) {
      return;
    }
    Properties props = transport.toDSProperties();
    if (displayName != null && !displayName.isEmpty()) {
      props.setProperty("name", displayName);
    }

    synchronized (systemLock) {
      if (system != null) {
        system.disconnect();
        system = null;
      }
      system = distributedSystemFactory.apply(props);
    }

    DistributionManager dm = system.getDistributionManager();
    if (dm instanceof ClusterDistributionManager) {
      ((ClusterDistributionManager) dm).setAgent(this);
    }

    synchronized (this) {
      disconnected = false;
    }

    system.addDisconnectListener(new InternalDistributedSystem.DisconnectListener() {
      @Override
      public String toString() {
        return String.format("Disconnect listener for %s", RemoteGfManagerAgent.this);
      }

      @Override
      public void onDisconnect(InternalDistributedSystem sys) {
        // Before the disconnect handler is called, the InternalDistributedSystem has already marked
        // itself for the disconnection. Hence the check for RemoteGfManagerAgent.this.isConnected()
        // always returns false. Hence commenting the same.
        boolean reconnect = sys.isReconnecting();
        if (!reconnect) {
          DisconnectListener listener = getDisconnectListener();
          if (disconnect()) {
            // returns true if disconnected during this call
            if (listener != null) {
              listener.onDisconnect(sys);
            }
          }
        }
      }
    });
    InternalDistributedSystem.addReconnectListener(new ReconnectListener() {
      @Override
      public void reconnecting(InternalDistributedSystem oldsys) {
        // nothing
      }

      @Override
      public void onReconnect(InternalDistributedSystem oldsys, InternalDistributedSystem newsys) {
        if (logger.isDebugEnabled()) {
          logger
              .debug("RemoteGfManagerAgent.onReconnect attempting to join new distributed system");
        }
        join();
      }
    });

    synchronized (myMembershipListenerLock) {
      myMembershipListener = myMembershipListenerFactory.apply(this);
      dm.addMembershipListener(myMembershipListener);
      Set<InternalDistributedMember> initialMembers = dm.getDistributionManagerIds();
      myMembershipListener.addMembers(new HashSet<>(initialMembers));

      if (logger.isDebugEnabled()) {
        StringBuilder sb = new StringBuilder("[RemoteGfManagerAgent] ");
        sb.append("Connected to DS with ");
        sb.append(initialMembers.size());
        sb.append(" members: ");

        for (InternalDistributedMember member : initialMembers) {
          sb.append(member);
          sb.append(" ");
        }

        logger.debug(sb.toString());
      }

      connected = true;

      for (InternalDistributedMember member : initialMembers) {
        // Create the admin objects synchronously. We don't need to use
        // the JoinProcess when we first start up.
        try {
          handleJoined(member);
        } catch (OperationCancelledException ex) {
          if (logger.isTraceEnabled(DM_VERBOSE)) {
            logger.trace(DM_VERBOSE, "join cancelled by departure");
          }
        }
      }

      initialized = true;
    }
  }
Creates a new {@link InternalDistributedSystem} connection for this agent. If one alread
exists, it is {@code disconnected} and a new one is created.
geode-core/src/main/java/org/apache/geode/internal/admin/remote/RemoteGfManagerAgent.java