@Override
  public GatewaySender create(String id, int remoteDSId) {
    int myDSId = InternalDistributedSystem.getAnyInstance().getDistributionManager()
        .getDistributedSystemId();
    if (remoteDSId == myDSId) {
      throw new GatewaySenderException(
          String.format(
              "GatewaySender %s cannot be created with remote DS Id equal to this DS Id. ",
              id));
    }
    if (remoteDSId < 0) {
      throw new GatewaySenderException(
          String.format("GatewaySender %s cannot be created with remote DS Id less than 0. ",
              id));
    }
    attrs.setId(id);
    attrs.setRemoteDs(remoteDSId);
    GatewaySender sender = null;

    if (attrs.getDispatcherThreads() <= 0) {
      throw new GatewaySenderException(
          String.format("GatewaySender %s can not be created with dispatcher threads less than 1",
              id));
    }
    // Verify socket read timeout if a proper logger is available
    if (cache instanceof GemFireCacheImpl) {
      // If socket read timeout is less than the minimum, log a warning.
      // Ideally, this should throw a GatewaySenderException, but wan dunit tests
      // were failing, and we were running out of time to change them.
      if (attrs.getSocketReadTimeout() != 0
          && attrs.getSocketReadTimeout() < GatewaySender.MINIMUM_SOCKET_READ_TIMEOUT) {
        logger.warn(
            "{} cannot configure socket read timeout of {} milliseconds because it is less than the minimum of {} milliseconds. The default will be used instead.",
            new Object[] {"GatewaySender " + id, attrs.getSocketReadTimeout(),
                GatewaySender.MINIMUM_SOCKET_READ_TIMEOUT});
        attrs.setSocketReadTimeout(GatewaySender.MINIMUM_SOCKET_READ_TIMEOUT);
      }

      if (attrs.getDiskStoreName() != null
          && cache.findDiskStore(attrs.getDiskStoreName()) == null) {
        throw new IllegalStateException(
            String.format("Disk store %s not found",
                attrs.getDiskStoreName()));
      }

      // Log a warning if the old system property is set.
      if (GATEWAY_CONNECTION_READ_TIMEOUT_PROPERTY_CHECKED.compareAndSet(false, true)) {
        if (System.getProperty(GatewaySender.GATEWAY_CONNECTION_READ_TIMEOUT_PROPERTY) != null) {
          logger.warn(
              "Obsolete java system property named {} was set to control {}. This property is no longer supported. Please use the GemFire API instead.",
              new Object[] {GatewaySender.GATEWAY_CONNECTION_READ_TIMEOUT_PROPERTY,
                  "GatewaySender socket read timeout"});
        }
      }
    }
    if (attrs.mustGroupTransactionEvents() && attrs.isBatchConflationEnabled()) {
      throw new GatewaySenderException(
          String.format(
              "GatewaySender %s cannot be created with both group transaction events set to true and batch conflation enabled",
              id));
    }

    if (attrs.isParallel()) {
      if ((attrs.getOrderPolicy() != null)
          && attrs.getOrderPolicy().equals(OrderPolicy.THREAD)) {
        throw new GatewaySenderException(
            String.format("Parallel Gateway Sender %s can not be created with OrderPolicy %s",
                id, attrs.getOrderPolicy()));
      }
      if (cache instanceof GemFireCacheImpl) {
        sender = new ParallelGatewaySenderImpl(cache, statisticsClock, attrs);
        cache.addGatewaySender(sender);

        if (!attrs.isManualStart()) {
          sender.start();
        }
      } else if (cache instanceof CacheCreation) {
        sender = new ParallelGatewaySenderCreation(cache, attrs);
        cache.addGatewaySender(sender);
      }
    } else {
      if (attrs.getAsyncEventListeners().size() > 0) {
        throw new GatewaySenderException(
            String.format(
                "SerialGatewaySender %s cannot define a remote site because at least AsyncEventListener is already added. Both listeners and remote site cannot be defined for the same gateway sender.",
                id));
      }
      if (attrs.mustGroupTransactionEvents() && attrs.getDispatcherThreads() > 1) {
        throw new GatewaySenderException(
            String.format(
                "SerialGatewaySender %s cannot be created with group transaction events set to true when dispatcher threads is greater than 1",
                id));
      }
      if (attrs.getOrderPolicy() == null && attrs.getDispatcherThreads() > 1) {
        attrs.setOrderPolicy(GatewaySender.DEFAULT_ORDER_POLICY);
      }
      if (cache instanceof GemFireCacheImpl) {
        sender = new SerialGatewaySenderImpl(cache, statisticsClock, attrs);
        cache.addGatewaySender(sender);
        if (!attrs.isManualStart()) {
          sender.start();
        }
      } else if (cache instanceof CacheCreation) {
        sender = new SerialGatewaySenderCreation(cache, attrs);
        cache.addGatewaySender(sender);
      }
    }
    return sender;
  }
Used internally to pass the attributes from this factory to the real GatewaySender it is
creating.
geode-wan/src/main/java/org/apache/geode/cache/wan/internal/GatewaySenderFactoryImpl.java