public static DataSerializer _register(DataSerializer s, boolean distribute) {
    final int id = s.getId();
    DataSerializer dsForMarkers = s;
    if (id == 0) {
      throw new IllegalArgumentException(
          "Cannot create a DataSerializer with id 0.");
    }
    final Class[] classes = s.getSupportedClasses();
    if (classes == null || classes.length == 0) {
      final String msg =
          "The DataSerializer %s has no supported classes. It's getSupportedClasses method must return at least one class";
      throw new IllegalArgumentException(String.format(msg, s.getClass().getName()));
    }

    for (Class aClass : classes) {
      if (aClass == null) {
        final String msg =
            "The DataSerializer getSupportedClasses method for %s returned an array that contained a null element.";
        throw new IllegalArgumentException(String.format(msg, s.getClass().getName()));
      } else if (aClass.isArray()) {
        final String msg =
            "The DataSerializer getSupportedClasses method for %s returned an array that contained an array class which is not allowed since arrays have built-in support.";
        throw new IllegalArgumentException(String.format(msg, s.getClass().getName()));
      }
    }

    final Integer idx = id;
    boolean retry;
    Marker oldMarker = null;
    final Marker m = new InitMarker();
    do {
      retry = false;
      Object oldSerializer = idsToSerializers.putIfAbsent(idx, m);
      if (oldSerializer != null) {
        if (oldSerializer instanceof Marker) {
          retry = !idsToSerializers.replace(idx, oldSerializer, m);
          if (!retry) {
            oldMarker = (Marker) oldSerializer;
          }
        } else if (oldSerializer.getClass().equals(s.getClass())) {
          // We've already got one of these registered
          if (distribute) {
            sendRegistrationMessage(s);
          }
          return (DataSerializer) oldSerializer;
        } else {
          DataSerializer other = (DataSerializer) oldSerializer;
          throw new IllegalStateException(
              String.format(
                  "A DataSerializer of class %s is already registered with id %s.",
                  other.getClass().getName(), other.getId()));
        }
      }
    } while (retry);

    try {
      for (int i = 0; i < classes.length; i++) {
        DataSerializer oldS = classesToSerializers.putIfAbsent(classes[i].getName(), s);
        if (oldS != null) {
          if (!s.equals(oldS)) {
            // cleanup the ones we have already added
            for (int j = 0; j < i; j++) {
              classesToSerializers.remove(classes[j].getName(), s);
            }
            dsForMarkers = null;
            String oldMsg;
            if (oldS.getId() == 0) {
              oldMsg = "DataSerializer has built-in support for class ";
            } else {
              oldMsg = "A DataSerializer of class " + oldS.getClass().getName()
                  + " is already registered to support class ";
            }
            String msg = oldMsg + classes[i].getName() + " so the DataSerializer of class "
                + s.getClass().getName() + " could not be registered.";
            if (oldS.getId() == 0) {
              throw new IllegalArgumentException(msg);
            } else {
              throw new IllegalStateException(msg);
            }
          }
        }
      }
    } finally {
      if (dsForMarkers == null) {
        idsToSerializers.remove(idx, m);
      } else {
        idsToSerializers.replace(idx, m, dsForMarkers);
      }
      if (oldMarker != null) {
        oldMarker.setSerializer(dsForMarkers);
      }
      m.setSerializer(dsForMarkers);
    }

    // if dataserializer is getting registered for first time
    // its EventID will be null, so generate a new event id
    // the the distributed system is connected
    InternalCache cache = getInternalCache();
    if (cache != null && s.getEventId() == null) {
      s.setEventId(new EventID(cache.getDistributedSystem()));
    }

    if (distribute) {
      // send a message to other peers telling them about a newly-registered
      // dataserializer, it also send event id of the originator along with the
      // dataserializer
      sendRegistrationMessage(s);
      // send it to cache servers if it is a client
      sendRegistrationMessageToServers(s);
    }
    // send it to all cache clients irrelevant of distribute
    // cache servers send it all the clients irrelevant of
    // originator VM
    sendRegistrationMessageToClients(s);

    fireNewDataSerializer(s);

    return s;
  }
Registers a {@code DataSerializer} instance with the data serialization framework.

@param distribute Should the registered {@code DataSerializer} be distributed to other members
       of the distributed system?
@see DataSerializer#register(Class)
geode-core/src/main/java/org/apache/geode/internal/InternalDataSerializer.java