static ClusterDistributionManager create(InternalDistributedSystem system,
      final MembershipLocator<InternalDistributedMember> membershipLocator) {

    ClusterDistributionManager distributionManager = null;
    boolean beforeJoined = true;

    try {

      int vmKind;

      if (Boolean.getBoolean(InternalLocator.FORCE_LOCATOR_DM_TYPE)) {
        // if this DM is starting for a locator, set it to be a locator DM
        vmKind = LOCATOR_DM_TYPE;

      } else if (isDedicatedAdminVM()) {
        vmKind = ADMIN_ONLY_DM_TYPE;

      } else {
        vmKind = NORMAL_DM_TYPE;
      }

      RemoteTransportConfig transport = new RemoteTransportConfig(system.getConfig(), vmKind);
      transport.setIsReconnectingDS(system.isReconnectingDS());
      transport.setOldDSMembershipInfo(system.oldDSMembershipInfo());

      long start = System.currentTimeMillis();

      distributionManager =
          new ClusterDistributionManager(system, transport, system.getAlertingService(),
              membershipLocator);
      distributionManager.assertDistributionManagerType();

      beforeJoined = false; // we have now joined the system

      {
        InternalDistributedMember id = distributionManager.getDistributionManagerId();
        if (!"".equals(id.getName())) {
          for (InternalDistributedMember m : distributionManager
              .getViewMembers()) {
            if (m.equals(id)) {
              // I'm counting on the members returned by getViewMembers being ordered such that
              // members that joined before us will precede us AND members that join after us
              // will succeed us.
              // SO once we find ourselves break out of this loop.
              break;
            }
            if (id.getName().equals(m.getName())) {
              if (distributionManager.getDistribution().verifyMember(m,
                  "member is using the name of " + id)) {
                throw new IncompatibleSystemException("Member " + id
                    + " could not join this distributed system because the existing member " + m
                    + " used the same name. Set the \"name\" gemfire property to a unique value.");
              }
            }
          }
        }
        distributionManager.addNewMember(id); // add ourselves
      }

      // Send out a StartupMessage to the other members.
      StartupOperation op = new StartupOperation(distributionManager, transport);

      try {
        if (!distributionManager.sendStartupMessage(op)) {
          // Well we didn't hear back from anyone else. We assume that
          // we're the first one.
          if (distributionManager.getOtherDistributionManagerIds().size() == 0) {
            logger.info("Did not hear back from any other system. I am the first one.");
          } else if (transport.isMcastEnabled()) {
            // perform a multicast ping test
            if (!distributionManager.testMulticast()) {
              logger.warn(
                  "Did not receive a startup response but other members exist.  Multicast does not seem to be working.");
            }
          }
        }
      } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
        // This is ALWAYS bad; don't consult a CancelCriterion.
        throw new InternalGemFireException(
            "Interrupted while waiting for first StartupResponseMessage",
            ex);
      } catch (IncompatibleSystemException ex) {
        logger.fatal(ex.getMessage(), ex);
        throw ex;
      } finally {
        distributionManager.readyToSendMsgs();
      }

      if (logger.isInfoEnabled()) {
        long delta = System.currentTimeMillis() - start;
        logger.info(LogMarker.DM_MARKER,
            "DistributionManager {} started on {}. There were {} other DMs. others: {} {} {}",
            distributionManager.getDistributionManagerId(), transport,
            distributionManager.getOtherDistributionManagerIds().size(),
            distributionManager.getOtherDistributionManagerIds(),
            (logger.isInfoEnabled(LogMarker.DM_MARKER) ? " (took " + delta + " ms)" : ""),
            ((distributionManager.getDMType() == ADMIN_ONLY_DM_TYPE) ? " (admin only)"
                : (distributionManager.getDMType() == LOCATOR_DM_TYPE) ? " (locator)" : ""));

        MembershipLogger.logStartup(distributionManager.getDistributionManagerId());
      }
      return distributionManager;
    } catch (RuntimeException r) {
      if (distributionManager != null) {
        if (logger.isDebugEnabled()) {
          logger.debug("cleaning up incompletely started DistributionManager due to exception", r);
        }
        distributionManager.uncleanShutdown(beforeJoined);
      }
      throw r;
    }
  }
Creates a new distribution manager and discovers the other members of the distributed system.
Note that it does not check to see whether or not this VM already has a distribution manager.

@param system The distributed system to which this distribution manager will send messages.
geode-core/src/main/java/org/apache/geode/distributed/internal/ClusterDistributionManager.java