@Override
  public Object fromBytes(byte[] bytes) {
    try {
      final Object object;
      final MessageType type;
      try {
        type = MessageType.values()[bytes[0]];
      } catch (ArrayIndexOutOfBoundsException e) {
        // The message type was introduced in samza 0.13.1. For samza 0.13.0 or older versions, the first byte of
        // MessageType doesn't exist in the bytes. Thus, upgrading from those versions will get this exception.
        // There are three ways to solve this issue:
        // a) Reset checkpoint to consume from newest message in the intermediate stream
        // b) clean all existing messages in the intermediate stream
        // c) Run the application in any version between 0.13.1 and 1.5 until all old messages in intermediate stream
        // has reached retention time.
        throw new SamzaException("Error reading the message type from intermediate message. This may happen if you "
            + "have recently upgraded from samza version older than 0.13.1 or there are still old messages in the "
            + "intermediate stream.", e);
      }
      final byte[] data = Arrays.copyOfRange(bytes, 1, bytes.length);
      switch (type) {
        case USER_MESSAGE:
          object = userMessageSerde.fromBytes(data);
          break;
        case WATERMARK:
          object = watermarkSerde.fromBytes(data);
          break;
        case END_OF_STREAM:
          object = eosSerde.fromBytes(data);
          break;
        case DRAIN:
          object = drainMessageSerde.fromBytes(data);
          break;
        default:
          throw new UnsupportedOperationException(String.format("Message type %s is not supported", type.name()));
      }
      return object;
    } catch (UnsupportedOperationException ue) {
      throw new SamzaException(ue);
    } catch (Exception e) {
      throw e;
    }
  }
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