private void doSendWriteSingleMessageTest(ByteBuffer encryptionKey, ByteBuffer expectedEncryptionKey,
      boolean verifyCrcError) throws Exception {
    String serviceId = "serviceId";
    String ownerId = "owner";
    String contentType = "bin";
    short accountId = 10;
    short containerId = 2;

    byte[] blob = TestUtils.getRandomBytes(10000);
    byte[] userMetadata = TestUtils.getRandomBytes(2000);
    StoreKey storeKey = new MockId("012345678910123456789012");
    BlobProperties properties =
        new BlobProperties(blob.length, serviceId, ownerId, contentType, false, 100, accountId, containerId,
            encryptionKey != null, null, null, null);
    MessageFormatInputStream putStream;
    MessageFormatRecord.MessageHeader_Format header;
    if (blobFormatVersion >= MessageFormatRecord.Blob_Version_V2) {
      header = getHeader(
          new PutMessageFormatInputStream(storeKey, encryptionKey == null ? null : encryptionKey.duplicate(),
              properties, ByteBuffer.wrap(userMetadata), new ByteBufferInputStream(ByteBuffer.wrap(blob)), blob.length,
              BlobType.DataBlob));
      putStream = new PutMessageFormatInputStream(storeKey, encryptionKey, properties, ByteBuffer.wrap(userMetadata),
          new ByteBufferInputStream(ByteBuffer.wrap(blob)), blob.length, BlobType.DataBlob);
    } else {
      header = getHeader(new PutMessageFormatBlobV1InputStream(storeKey, properties, ByteBuffer.wrap(userMetadata),
          new ByteBufferInputStream(ByteBuffer.wrap(blob)), blob.length, BlobType.DataBlob));
      putStream = new PutMessageFormatBlobV1InputStream(storeKey, properties, ByteBuffer.wrap(userMetadata),
          new ByteBufferInputStream(ByteBuffer.wrap(blob)), blob.length, BlobType.DataBlob);
    }

    ByteBuffer buf1 = ByteBuffer.allocate((int) putStream.getSize());

    putStream.read(buf1.array());
    if (verifyCrcError) {
      // The last 8 bytes are the crc for the blob format, let's change some bytes before, and if the MessageFormatSend
      // would be able to catch that.
      byte[] blobAll = buf1.array();
      // this is the last byte of the blob format content
      byte b = blobAll[blobAll.length - MessageFormatRecord.Crc_Size - 1];
      b = b == 0xff ? (byte) 0x00 : (byte) 0xff;
      blobAll[blobAll.length - MessageFormatRecord.Crc_Size - 1] = b;
    }

    List<ByteBuffer> listbuf = Arrays.asList(buf1);
    List<StoreKey> storeKeys = Arrays.asList(storeKey);
    MockMessageReadSet readSet = new MockMessageReadSet(listbuf, storeKeys);

    MessageFormatMetrics metrics = new MessageFormatMetrics(new MetricRegistry());
    long exceptionCountBefore = metrics.messageFormatExceptionCount.getCount();
    long exceptionCountAfter = exceptionCountBefore;
    // get all
    MessageFormatSend send = new MessageFormatSend(readSet, MessageFormatFlags.All, metrics, new MockIdFactory());
    Assert.assertEquals(send.sizeInBytes(), putStream.getSize());
    Assert.assertEquals(1, send.getMessageMetadataList().size());
    Assert.assertEquals(null, send.getMessageMetadataList().get(0));
    ByteBuffer bufresult = ByteBuffer.allocate((int) putStream.getSize());
    WritableByteChannel channel = Channels.newChannel(new ByteBufferOutputStream(bufresult));
    while (!send.isSendComplete()) {
      send.writeTo(channel);
    }
    Assert.assertArrayEquals(buf1.array(), bufresult.array());
    Assert.assertTrue(readSet.isPrefetchInfoCorrect(0, readSet.sizeInBytes(0)));
    if (verifyCrcError) {
      exceptionCountAfter = metrics.messageFormatExceptionCount.getCount();
      Assert.assertEquals(exceptionCountBefore + 1, exceptionCountAfter);
      exceptionCountBefore = exceptionCountAfter;
    }

    // get blob
    send = new MessageFormatSend(readSet, MessageFormatFlags.Blob, metrics, new MockIdFactory());
    long blobRecordSize = (blobFormatVersion == MessageFormatRecord.Blob_Version_V3)
        ? MessageFormatRecord.Blob_Format_V3.getBlobRecordSize(blob.length)
        : ((blobFormatVersion == MessageFormatRecord.Blob_Version_V2)
            ? MessageFormatRecord.Blob_Format_V2.getBlobRecordSize(blob.length)
            : MessageFormatRecord.Blob_Format_V1.getBlobRecordSize(blob.length));
    Assert.assertEquals(send.sizeInBytes(), blobRecordSize);
    bufresult.clear();
    channel = Channels.newChannel(new ByteBufferOutputStream(bufresult));
    while (!send.isSendComplete()) {
      send.writeTo(channel);
    }

    int toIndex = verifyCrcError ? blob.length - 1 : blob.length;
    for (int i = 0; i < toIndex; i++) {
      Assert.assertEquals(blob[i],
          bufresult.array()[i + (int) blobRecordSize - MessageFormatRecord.Crc_Size - blob.length]);
    }

    if (expectedEncryptionKey == null) {
      Assert.assertEquals(null, send.getMessageMetadataList().get(0));
    } else {
      Assert.assertEquals(expectedEncryptionKey, send.getMessageMetadataList().get(0).getEncryptionKey());
    }
    Assert.assertTrue(readSet.isPrefetchInfoCorrect(0, readSet.sizeInBytes(0)));
    if (verifyCrcError) {
      exceptionCountAfter = metrics.messageFormatExceptionCount.getCount();
      Assert.assertEquals(exceptionCountBefore + 1, exceptionCountAfter);
    }

    // get user metadata
    send = new MessageFormatSend(readSet, MessageFormatFlags.BlobUserMetadata, metrics, new MockIdFactory());
    long userMetadataRecordSize =
        MessageFormatRecord.UserMetadata_Format_V1.getUserMetadataSize(ByteBuffer.wrap(userMetadata));
    Assert.assertEquals(send.sizeInBytes(), userMetadataRecordSize);
    bufresult.clear();
    channel = Channels.newChannel(new ByteBufferOutputStream(bufresult));
    while (!send.isSendComplete()) {
      send.writeTo(channel);
    }

    bufresult.flip();
    // read off the header.
    for (int i = 0; i < userMetadataRecordSize - MessageFormatRecord.Crc_Size - userMetadata.length; i++) {
      bufresult.get();
    }

    verifyBlobUserMetadata(userMetadata, bufresult);
    if (expectedEncryptionKey == null) {
      Assert.assertEquals(null, send.getMessageMetadataList().get(0));
    } else {
      Assert.assertEquals(expectedEncryptionKey, send.getMessageMetadataList().get(0).getEncryptionKey());
    }
    Assert.assertTrue(readSet.isPrefetchInfoCorrect(header.getUserMetadataRecordRelativeOffset(),
        header.getUserMetadataRecordSize()));

    // get blob properties
    send = new MessageFormatSend(readSet, MessageFormatFlags.BlobProperties, metrics, new MockIdFactory());
    long blobPropertiesRecordSize =
        MessageFormatRecord.BlobProperties_Format_V1.getBlobPropertiesRecordSize(properties);
    Assert.assertEquals(send.sizeInBytes(), blobPropertiesRecordSize);
    bufresult.clear();
    channel = Channels.newChannel(new ByteBufferOutputStream(bufresult));
    while (!send.isSendComplete()) {
      send.writeTo(channel);
    }

    bufresult.flip();
    // read off the header.
    for (int i = 0;
        i < blobPropertiesRecordSize - MessageFormatRecord.Crc_Size - BlobPropertiesSerDe.getBlobPropertiesSerDeSize(
            properties); i++) {
      bufresult.get();
    }

    verifyBlobProperties(properties,
        BlobPropertiesSerDe.getBlobPropertiesFromStream(new DataInputStream(new ByteBufferInputStream(bufresult))));
    Assert.assertEquals(null, send.getMessageMetadataList().get(0));
    Assert.assertTrue(readSet.isPrefetchInfoCorrect(header.getBlobPropertiesRecordRelativeOffset(),
        header.getBlobPropertiesRecordSize()));

    // get blob info
    send = new MessageFormatSend(readSet, MessageFormatFlags.BlobInfo, metrics, new MockIdFactory());
    Assert.assertEquals(send.sizeInBytes(), blobPropertiesRecordSize + userMetadataRecordSize);
    bufresult.clear();
    channel = Channels.newChannel(new ByteBufferOutputStream(bufresult));
    while (!send.isSendComplete()) {
      send.writeTo(channel);
    }

    bufresult.flip();
    for (int i = 0;
        i < blobPropertiesRecordSize - MessageFormatRecord.Crc_Size - BlobPropertiesSerDe.getBlobPropertiesSerDeSize(
            properties); i++) {
      bufresult.get();
    }
    verifyBlobProperties(properties,
        BlobPropertiesSerDe.getBlobPropertiesFromStream(new DataInputStream(new ByteBufferInputStream(bufresult))));
    for (int i = 0; i < userMetadataRecordSize - userMetadata.length; i++) {
      bufresult.get();
    }
    verifyBlobUserMetadata(userMetadata, bufresult);
    if (expectedEncryptionKey == null) {
      Assert.assertEquals(null, send.getMessageMetadataList().get(0));
    } else {
      Assert.assertEquals(expectedEncryptionKey, send.getMessageMetadataList().get(0).getEncryptionKey());
    }
    Assert.assertTrue(readSet.isPrefetchInfoCorrect(header.getBlobPropertiesRecordRelativeOffset(),
        header.getBlobPropertiesRecordSize() + header.getUserMetadataRecordSize()));
  }
Helper method for testing single message sends.
@param encryptionKey the encryption key to include in the message while writing it.
@param expectedEncryptionKey the key expected when reading the sent message.
@param verifyCrcError True to change blob content so the crc would be mismatch
ambry-messageformat/src/test/java/com/github/ambry/messageformat/MessageFormatSendTest.java