private void initializeRegion(AbstractGatewaySender sender,
      CacheListener<Long, AsyncEvent<?, ?>> listener) {
    final InternalCache gemCache = sender.getCache();
    region = gemCache.getRegion(regionName);
    if (region == null) {
      RegionShortcut regionShortcut;
      if (enablePersistence) {
        regionShortcut = RegionShortcut.REPLICATE_PERSISTENT;
      } else {
        regionShortcut = RegionShortcut.REPLICATE;
      }
      InternalRegionFactory<Long, AsyncEvent<?, ?>> factory =
          gemCache.createInternalRegionFactory(regionShortcut);
      if (NO_ACK) {
        factory.setScope(Scope.DISTRIBUTED_NO_ACK);
      }
      if (logger.isDebugEnabled()) {
        logger.debug("The policy of region is {}",
            (enablePersistence ? DataPolicy.PERSISTENT_REPLICATE : DataPolicy.REPLICATE));
      }
      // Set listener if it is not null. The listener will be non-null
      // when the user of this queue is a secondary VM.
      if (listener != null) {
        factory.addCacheListener(listener);
      }
      // allow for no overflow directory
      EvictionAttributes ea = EvictionAttributes.createLIFOMemoryAttributes(maximumQueueMemory,
          EvictionAction.OVERFLOW_TO_DISK);

      factory.setEvictionAttributes(ea);
      factory.setConcurrencyChecksEnabled(false);

      factory.setDiskStoreName(diskStoreName);

      // In case of persistence write to disk sync and in case of eviction write in async
      factory.setDiskSynchronous(isDiskSynchronous);

      // Create the region
      if (logger.isDebugEnabled()) {
        logger.debug("{}: Attempting to create queue region: {}", this, regionName);
      }
      final RegionAttributes<Long, AsyncEvent<?, ?>> ra = factory.getCreateAttributes();
      try {
        SerialGatewaySenderQueueMetaRegion meta =
            metaRegionFactory.newMetaRegion(gemCache, regionName, ra, sender);
        factory
            .setInternalMetaRegion(meta)
            .setDestroyLockFlag(true)
            .setSnapshotInputStream(null)
            .setImageTarget(null)
            .setIsUsedForSerialGatewaySenderQueue(true)
            .setInternalRegion(true)
            .setSerialGatewaySender(sender);
        region = factory.create(regionName);
        // Add overflow statistics to the mbean
        addOverflowStatisticsToMBean(gemCache, sender);

        if (logger.isDebugEnabled()) {
          logger.debug("{}: Created queue region: {}", this, region);
        }
      } catch (CacheException e) {
        logger.fatal(String.format("%s: The queue region named %s could not be created",
            this, regionName),
            e);
      }
    } else {
      if (listener != null) {
        addCacheListener(listener);
      }
    }
    if ((region != null) && cleanQueues) {
      region.clear();
    }
  }
Initializes the <code>Region</code> backing this queue. The <code>Region</code>'s scope is
DISTRIBUTED_NO_ACK and mirror type is KEYS_VALUES and is set to overflow to disk based on the
<code>GatewayQueueAttributes</code>.

@param sender The GatewaySender <code>SerialGatewaySenderImpl</code>
@param listener The GemFire <code>CacheListener</code>. The <code>CacheListener</code> can be
       null.
geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderQueue.java