private Message newMessage(InputStream inputStream, StoreKey newKey, MessageInfo oldMessageInfo) throws Exception {
    MessageHeader_Format headerFormat = getMessageHeader(inputStream);
    storeKeyFactory.getStoreKey(new DataInputStream(inputStream));
    BlobId newBlobId = (BlobId) newKey;

    if (headerFormat.isPutRecord()) {
      if (headerFormat.hasLifeVersion() && headerFormat.getLifeVersion() != oldMessageInfo.getLifeVersion()) {
        // The original Put buffer might have lifeVersion as 0, but the message info might have a higher lifeVersion.
        logger.trace("LifeVersion in stream: {} failed to match lifeVersion from Index: {} for key {}",
            headerFormat.getLifeVersion(), oldMessageInfo.getLifeVersion(), oldMessageInfo.getStoreKey());
      }
      ByteBuffer blobEncryptionKey = null;
      if (headerFormat.hasEncryptionKeyRecord()) {
        blobEncryptionKey = deserializeBlobEncryptionKey(inputStream);
      }
      BlobProperties oldProperties = deserializeBlobProperties(inputStream);
      ByteBuffer userMetaData = deserializeUserMetadata(inputStream);
      BlobData blobData = deserializeBlob(inputStream);
      ByteBuf blobDataBytes = blobData.content();

      long blobPropertiesSize = oldProperties.getBlobSize();

      //If the blob is a metadata blob its data chunk id list
      //will be rewritten with transformed IDs
      if (blobData.getBlobType().equals(BlobType.MetadataBlob)) {
        ByteBuffer serializedMetadataContent = blobDataBytes.nioBuffer();
        CompositeBlobInfo compositeBlobInfo =
            MetadataContentSerDe.deserializeMetadataContentRecord(serializedMetadataContent, storeKeyFactory);
        Map<StoreKey, StoreKey> convertedKeys = storeKeyConverter.convert(compositeBlobInfo.getKeys());
        List<StoreKey> newKeys = new ArrayList<>();
        boolean isOldMetadataKeyDifferentFromNew = !oldMessageInfo.getStoreKey().getID().equals(newKey.getID());
        short metadataAccountId = newBlobId.getAccountId();
        short metadataContainerId = newBlobId.getContainerId();
        for (StoreKey oldDataChunkKey : compositeBlobInfo.getKeys()) {
          StoreKey newDataChunkKey = convertedKeys.get(oldDataChunkKey);
          if (newDataChunkKey == null) {
            throw new IllegalStateException("Found metadata chunk with a deprecated data chunk. " + " Old MetadataID: "
                + oldMessageInfo.getStoreKey().getID() + " New MetadataID: " + newKey.getID() + " Old Datachunk ID: "
                + oldDataChunkKey.getID());
          }
          if (isOldMetadataKeyDifferentFromNew && newDataChunkKey.getID().equals(oldDataChunkKey.getID())) {
            throw new IllegalStateException(
                "Found changed metadata chunk with an unchanged data chunk" + " Old MetadataID: "
                    + oldMessageInfo.getStoreKey().getID() + " New MetadataID: " + newKey.getID()
                    + " Old Datachunk ID: " + oldDataChunkKey.getID());
          }
          if (!isOldMetadataKeyDifferentFromNew && !newDataChunkKey.getID().equals(oldDataChunkKey.getID())) {
            throw new IllegalStateException(
                "Found unchanged metadata chunk with a changed data chunk" + " Old MetadataID: "
                    + oldMessageInfo.getStoreKey().getID() + " New MetadataID: " + newKey.getID()
                    + " Old Datachunk ID: " + oldDataChunkKey.getID() + " New Datachunk ID: "
                    + newDataChunkKey.getID());
          }
          BlobId newDataChunkBlobId = (BlobId) newDataChunkKey;
          if (newDataChunkBlobId.getAccountId() != metadataAccountId
              || newDataChunkBlobId.getContainerId() != metadataContainerId) {
            throw new IllegalStateException(
                "Found changed metadata chunk with a datachunk with a different account/container" + " Old MetadataID: "
                    + oldMessageInfo.getStoreKey().getID() + " New MetadataID: " + newKey.getID()
                    + " Old Datachunk ID: " + oldDataChunkKey.getID() + " New Datachunk ID: "
                    + newDataChunkBlobId.getID() + " Metadata AccountId: " + metadataAccountId
                    + " Metadata ContainerId: " + metadataContainerId + " Datachunk AccountId: "
                    + newDataChunkBlobId.getAccountId() + " Datachunk ContainerId: "
                    + newDataChunkBlobId.getContainerId());
          }
          newKeys.add(newDataChunkKey);
        }
        ByteBuffer metadataContent;
        if (compositeBlobInfo.getMetadataContentVersion() == Metadata_Content_Version_V2) {
          metadataContent = MetadataContentSerDe.serializeMetadataContentV2(compositeBlobInfo.getChunkSize(),
              compositeBlobInfo.getTotalSize(), newKeys);
        } else if (compositeBlobInfo.getMetadataContentVersion() == Metadata_Content_Version_V3) {
          List<Pair<StoreKey, Long>> keyAndSizeList = new ArrayList<>();
          List<CompositeBlobInfo.ChunkMetadata> chunkMetadataList = compositeBlobInfo.getChunkMetadataList();
          for (int i = 0; i < newKeys.size(); i++) {
            keyAndSizeList.add(new Pair<>(newKeys.get(i), chunkMetadataList.get(i).getSize()));
          }
          metadataContent =
              MetadataContentSerDe.serializeMetadataContentV3(compositeBlobInfo.getTotalSize(), keyAndSizeList);
        } else {
          throw new IllegalStateException("Unexpected metadata content version from composite blob: "
              + compositeBlobInfo.getMetadataContentVersion());
        }
        blobPropertiesSize = compositeBlobInfo.getTotalSize();
        metadataContent.flip();
        blobDataBytes.release();
        blobDataBytes = Unpooled.wrappedBuffer(metadataContent);
        blobData = new BlobData(blobData.getBlobType(), metadataContent.remaining(), blobDataBytes);
      }

      BlobProperties newProperties =
          new BlobProperties(blobPropertiesSize, oldProperties.getServiceId(), oldProperties.getOwnerId(),
              oldProperties.getContentType(), oldProperties.isPrivate(), oldProperties.getTimeToLiveInSeconds(),
              oldProperties.getCreationTimeInMs(), newBlobId.getAccountId(), newBlobId.getContainerId(),
              oldProperties.isEncrypted(), oldProperties.getExternalAssetTag(), oldProperties.getContentEncoding(),
              oldProperties.getFilename(), oldProperties.getReservedMetadataBlobId());

      // 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 putMessageFormatInputStream =
          new PutMessageFormatInputStream(newKey, blobEncryptionKey, newProperties, userMetaData,
              new ByteBufInputStream(blobDataBytes, true), blobData.getSize(), blobData.getBlobType(),
              oldMessageInfo.getLifeVersion(), blobData.isCompressed());
      // Reuse the original CRC if present in the oldMessageInfo. This is important to ensure that messages that are
      // received via replication are sent to the store with proper CRCs (which the store needs to detect duplicate
      // messages). As an additional guard, here the original CRC is only reused if the key's ID in string form is the
      // same after conversion.
      // set isDeleted same as the oldMessageInfo
      Long originalCrc = oldMessageInfo.getStoreKey().getID().equals(newKey.getID()) ? oldMessageInfo.getCrc() : null;
      MessageInfo info =
          new MessageInfo.Builder(newKey, putMessageFormatInputStream.getSize(), newProperties.getAccountId(),
              newProperties.getContainerId(), oldMessageInfo.getOperationTimeMs()).isTtlUpdated(
              oldMessageInfo.isTtlUpdated()).isDeleted(oldMessageInfo.isDeleted())
              .expirationTimeInMs(oldMessageInfo.getExpirationTimeInMs())
              .crc(originalCrc)
              .lifeVersion(oldMessageInfo.getLifeVersion())
              .build();
      return new Message(info, putMessageFormatInputStream);
    } else {
      throw new IllegalArgumentException("Only 'put' records are valid");
    }
  }
Creates a Message from the old Message
input stream, replacing the old store key and account/container IDs
with a new store key and account/container IDs
@param inputStream the input stream of the Message
@param newKey the new StoreKey
@param oldMessageInfo the {@link MessageInfo} of the message being transformed
@return new Message message
@throws Exception
ambry-replication/src/main/java/com/github/ambry/replication/BlobIdTransformer.java