@Override
  public void toData(DataOutput out,
      SerializationContext context) throws IOException {
    super.toData(out, context);

    HeapDataOutputStream hdos = null;
    boolean failedSerialization = false;
    final boolean hasReturnValue = returnValueIsException || returnValue != null;
    if (hasReturnValue) {
      hdos = new HeapDataOutputStream(context.getSerializationVersion());
      try {
        context.getSerializer().writeObject(returnValue, hdos);
      } catch (NotSerializableException e) {
        logger.warn("Unable to serialize a reply to " + getRecipientsDescription(), e);
        failedSerialization = true;
        returnValue = new ReplyException(e);
        returnValueIsException = true;
      }
    }
    byte status = 0;
    if (ignored) {
      status |= IGNORED_FLAG;
    }
    if (returnValueIsException) {
      status |= EXCEPTION_FLAG;
    } else if (returnValue != null) {
      status |= OBJECT_FLAG;
    }
    if (processorId != 0) {
      status |= PROCESSOR_ID_FLAG;
    }
    if (closed) {
      status |= CLOSED_FLAG;
    }
    if (internal) {
      status |= INTERNAL_FLAG;
    }
    out.writeByte(status);
    if (processorId != 0) {
      out.writeInt(processorId);
    }
    if (hasReturnValue) {
      if (failedSerialization) {
        context.getSerializer().writeObject(returnValue, out);
      } else {
        hdos.sendTo(out);
      }
    }
  }
Processes this message. This method is invoked by the receiver of the message if the message is
not direct ack. If the message is a direct ack, the process(dm, ReplyProcessor) method is
invoked instead.

@param dm the distribution manager that is processing the message.
geode-core/src/main/java/org/apache/geode/distributed/internal/ReplyMessage.java