private void createStreamWithMessageHeader(StoreKey key, ByteBuffer blobEncryptionKey, BlobProperties blobProperties,
      ByteBuffer userMetadata, InputStream blobStream, long streamSize, BlobType blobType, short lifeVersion,
      boolean isCompressed)
      throws MessageFormatException {
    int headerSize = MessageFormatRecord.getHeaderSizeForVersion(MessageFormatRecord.headerVersionToUse);
    int blobEncryptionKeySize = blobEncryptionKey == null ? 0
        : MessageFormatRecord.BlobEncryptionKey_Format_V1.getBlobEncryptionKeyRecordSize(blobEncryptionKey);
    int blobPropertiesRecordSize =
        MessageFormatRecord.BlobProperties_Format_V1.getBlobPropertiesRecordSize(blobProperties);
    int userMetadataSize = MessageFormatRecord.UserMetadata_Format_V1.getUserMetadataSize(userMetadata);
    long blobSize = MessageFormatRecord.Blob_Format_V3.getBlobRecordSize(streamSize);

    buffer = ByteBuffer.allocate(
        headerSize + key.sizeInBytes() + blobEncryptionKeySize + blobPropertiesRecordSize + userMetadataSize + (int) (
            blobSize - streamSize - MessageFormatRecord.Crc_Size));

    long totalSize = blobEncryptionKeySize + blobPropertiesRecordSize + userMetadataSize + blobSize;
    int blobEncryptionKeyRecordRelativeOffset =
        blobEncryptionKey == null ? MessageFormatRecord.Message_Header_Invalid_Relative_Offset
            : headerSize + key.sizeInBytes();
    int blobPropertiesRecordRelativeOffset = blobEncryptionKey == null ? headerSize + key.sizeInBytes()
        : blobEncryptionKeyRecordRelativeOffset + blobEncryptionKeySize;
    int updateRecordRelativeOffset = MessageFormatRecord.Message_Header_Invalid_Relative_Offset;
    int userMetadataRecordRelativeOffset = blobPropertiesRecordRelativeOffset + blobPropertiesRecordSize;
    int blobRecordRelativeOffset = userMetadataRecordRelativeOffset + userMetadataSize;
    if (MessageFormatRecord.headerVersionToUse == MessageFormatRecord.Message_Header_Version_V2) {
      MessageFormatRecord.MessageHeader_Format_V2.serializeHeader(buffer, totalSize,
          blobEncryptionKeyRecordRelativeOffset, blobPropertiesRecordRelativeOffset, updateRecordRelativeOffset,
          userMetadataRecordRelativeOffset, blobRecordRelativeOffset);
    } else {
      MessageFormatRecord.MessageHeader_Format_V3.serializeHeader(buffer, lifeVersion, totalSize,
          blobEncryptionKeyRecordRelativeOffset, blobPropertiesRecordRelativeOffset, updateRecordRelativeOffset,
          userMetadataRecordRelativeOffset, blobRecordRelativeOffset);
    }
    buffer.put(key.toBytes());
    if (blobEncryptionKey != null) {
      MessageFormatRecord.BlobEncryptionKey_Format_V1.serializeBlobEncryptionKeyRecord(buffer, blobEncryptionKey);
    }
    MessageFormatRecord.BlobProperties_Format_V1.serializeBlobPropertiesRecord(buffer, blobProperties);
    MessageFormatRecord.UserMetadata_Format_V1.serializeUserMetadataRecord(buffer, userMetadata);
    int bufferBlobStart = buffer.position();
    MessageFormatRecord.Blob_Format_V3.serializePartialBlobRecord(buffer, streamSize, blobType, isCompressed);
    CRC32 crc = new CRC32();
    crc.update(buffer.array(), bufferBlobStart, buffer.position() - bufferBlobStart);
    stream = new CrcInputStream(crc, blobStream);
    streamLength = streamSize;
    messageLength = buffer.capacity() + streamLength + MessageFormatRecord.Crc_Size;
    buffer.flip();
  }
Helper method to create a stream with encryption key record. This will be the standard once all nodes in a cluster
understand reading messages with encryption key record.
ambry-messageformat/src/main/java/com/github/ambry/messageformat/PutMessageFormatInputStream.java