@Override
  public byte[] toBytes(Object object) {
    final byte[] data;
    final MessageType type = MessageType.of(object);
    switch (type) {
      case USER_MESSAGE:
        data = userMessageSerde.toBytes(object);
        break;
      case WATERMARK:
        data = watermarkSerde.toBytes((WatermarkMessage) object);
        break;
      case END_OF_STREAM:
        data = eosSerde.toBytes((EndOfStreamMessage) object);
        break;
      case DRAIN:
        data = drainMessageSerde.toBytes((DrainMessage) object);
        break;
      default:
        throw new SamzaException("Unknown message type: " + type.name());
    }

    final byte[] bytes = new byte[data.length + 1];
    bytes[0] = (byte) type.ordinal();
    System.arraycopy(data, 0, bytes, 1, data.length);

    return bytes;
  }
This class provides serialization/deserialization of the intermediate messages.

The message format of an intermediate stream is below:

IntermediateStreamMessage: {
  MessageType : int8
  MessageData : byte[]
}

MessageType: [0(UserMessage), 1(Watermark), 2(EndOfStream)]
MessageData: [UserMessage/ControlMessage]
ControlMessage:
  Version   : int
  TaskName  : string
  TaskCount : int
  Other Message Data (based on different types of control message)

For user message, we use the user message serde.
For control message, we use json serde.
samza-core/src/main/java/org/apache/samza/serializers/IntermediateMessageSerde.java