@Override
  public TransformationOutput transform(Message message) {
    ByteBuffer encryptionKey;
    BlobProperties props;
    ByteBuffer metadata;
    BlobData blobData;
    MessageInfo msgInfo = message.getMessageInfo();
    InputStream msgStream = message.getStream();
    TransformationOutput transformationOutput = null;
    try {
      // Read header
      ByteBuffer headerVersion = ByteBuffer.allocate(Version_Field_Size_In_Bytes);
      msgStream.read(headerVersion.array());
      short version = headerVersion.getShort();
      if (!isValidHeaderVersion(version)) {
        throw new MessageFormatException("Header version not supported " + version,
            MessageFormatErrorCodes.DataCorrupt);
      }
      int headerSize = getHeaderSizeForVersion(version);
      ByteBuffer headerBuffer = ByteBuffer.allocate(headerSize);
      headerBuffer.put(headerVersion.array());
      msgStream.read(headerBuffer.array(), Version_Field_Size_In_Bytes, headerSize - Version_Field_Size_In_Bytes);
      headerBuffer.rewind();
      MessageHeader_Format header = getMessageHeader(version, headerBuffer);
      header.verifyHeader();
      StoreKey keyInStream = storeKeyFactory.getStoreKey(new DataInputStream(msgStream));
      if (header.isPutRecord()) {
        if (header.hasLifeVersion() && header.getLifeVersion() != msgInfo.getLifeVersion()) {
          logger.trace("LifeVersion in stream: {} failed to match lifeVersion from Index: {} for key {}",
              header.getLifeVersion(), msgInfo.getLifeVersion(), keyInStream);
        }
        encryptionKey = header.hasEncryptionKeyRecord() ? deserializeBlobEncryptionKey(msgStream) : null;
        props = deserializeBlobProperties(msgStream);
        metadata = deserializeUserMetadata(msgStream);
        blobData = deserializeBlob(msgStream);
      } else {
        throw new IllegalStateException("Message cannot be anything rather than put record ");
      }
      if (msgInfo.getStoreKey().equals(keyInStream)) {
        // BlobIDTransformer only exists on ambry-server and replication between servers is relying on blocking channel
        // which is still using java ByteBuffer. So, no need to consider releasing stuff.
        // @todo, when netty Bytebuf is adopted for blocking channel on ambry-server, remember to release this ByteBuf.
        PutMessageFormatInputStream transformedStream =
            new PutMessageFormatInputStream(keyInStream, encryptionKey, props, metadata,
                new ByteBufInputStream(blobData.content(), true), blobData.getSize(), blobData.getBlobType(),
                msgInfo.getLifeVersion(), blobData.isCompressed());
        MessageInfo transformedMsgInfo = new MessageInfo.Builder(msgInfo).size(transformedStream.getSize())
            .isDeleted(false)
            .isUndeleted(false)
            .build();
        transformationOutput = new TransformationOutput(new Message(transformedMsgInfo, transformedStream));
      } else {
        throw new IllegalStateException(
            "StoreKey in stream: " + keyInStream + " failed to match store key from Index: " + msgInfo.getStoreKey());
      }
    } catch (Exception e) {
      transformationOutput = new TransformationOutput(e);
    }
    return transformationOutput;
  }
An implementation of the {@link Transformer} interface that simply deserializes the message to validate it.
ambry-messageformat/src/main/java/com/github/ambry/messageformat/ValidatingTransformer.java