@Override
  public void initialize() {
    // Relevant during reconnect
    TypeRegistry typeRegistry = cache.getPdxRegistry();
    if (typeRegistry != null) {
      typeRegistry.flushCache();
      logger.debug("Flushing TypeRegistry");
    }

    InternalRegionFactory factory = cache.createInternalRegionFactory();

    factory.setScope(Scope.DISTRIBUTED_ACK);
    if (cache.getPdxPersistent()) {
      if (cache.getCacheConfig().pdxDiskStoreUserSet) {
        factory.setDiskStoreName(cache.getPdxDiskStore());
      } else {
        factory.setDiskStoreName(cache.getOrCreateDefaultDiskStore().getName());
      }
      factory.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE);
    } else {
      factory.setDataPolicy(DataPolicy.REPLICATE);
    }

    // Add a listener that makes sure that if anyone in the DS is using PDX
    // Our PDX configuration is valid for this member. This is important if
    // we are the gateway, we need to validate that we have a distributed system
    // id.
    factory.addCacheListener(new CacheListenerAdapter<Object, Object>() {
      @Override
      public void afterCreate(EntryEvent<Object, Object> event) {
        verifyConfiguration();
        // update the local map and reverse map with the pdxtypes registered
        Object value = event.getNewValue();
        Object key = event.getKey();
        if (value != null) {
          updateLocalAndReverseMaps(key, value);
        }
      }
    });

    factory.setCacheWriter(new CacheWriterAdapter<Object, Object>() {

      @Override
      public void beforeCreate(EntryEvent<Object, Object> event) throws CacheWriterException {
        Object newValue = event.getNewValue();
        if (newValue instanceof PdxType) {
          logger.info("Adding new type: {}", ((PdxType) event.getNewValue()).toFormattedString());
        } else {
          logger.info("Adding new type: {} {}", event.getKey(),
              ((EnumInfo) newValue).toFormattedString());
        }
      }

      @Override
      public void beforeUpdate(EntryEvent<Object, Object> event) throws CacheWriterException {
        if (!event.getRegion().get(event.getKey()).equals(event.getNewValue())) {
          PdxRegistryMismatchException ex = new PdxRegistryMismatchException(
              "Trying to add a PDXType with the same id as an existing PDX type. id="
                  + event.getKey() + ", existing pdx type " + event.getOldValue() + ", new type "
                  + event.getNewValue());
          throw new CacheWriterException(ex);
        }
      }

    });

    factory.setIsUsedForMetaRegion(true);
    factory.setMetaRegionWithTransactions(true);
    try {
      idToType = factory.create(REGION_NAME);
    } catch (TimeoutException | RegionExistsException ex) {
      throw new PdxInitializationException("Could not create pdx registry", ex);
    }

    statistics.initialize();

    // If there is anything in the id to type registry,
    // we should validate our configuration now.
    // And send those types to any existing gateways.
    if (!getIdToType().isEmpty()) {
      verifyConfiguration();
    }
  }
The region where the PDX metadata is stored. Because this region is transactional for our
internal updates but we don't want to participate in the users transactions, all operations on
this region must suspend any existing transactions with suspendTX/resumeTX.
geode-core/src/main/java/org/apache/geode/pdx/internal/PeerTypeRegistration.java