static MessageInfoAndMetadataListSerde deserializeMessageInfoAndMetadataList(DataInputStream stream, ClusterMap map,
      short versionToDeserializeIn) throws IOException {
    if (versionToDeserializeIn >= VERSION_5) {
      short versionFromStream = stream.readShort();
      if (versionFromStream != versionToDeserializeIn) {
        throw new IllegalArgumentException(
            "Argument provided [" + versionToDeserializeIn + "] and stream [" + versionFromStream
                + "] disagree on version");
      }
    } else {
      versionToDeserializeIn =
          versionToDeserializeIn == DETERMINE_VERSION ? stream.readShort() : versionToDeserializeIn;
    }
    int messageCount = stream.readInt();
    ArrayList<MessageInfo> messageInfoList = new ArrayList<>(messageCount);
    ArrayList<MessageMetadata> messageMetadataList = new ArrayList<>(messageCount);
    for (int i = 0; i < messageCount; i++) {
      BlobId id = new BlobId(stream, map);
      long size = stream.readLong();
      long ttl = stream.readLong();
      boolean isDeleted = stream.readByte() == UPDATED;
      boolean isTtlUpdated = false;
      boolean isUndeleted = false;
      short lifeVersion = 0;
      Long crc = null;
      short accountId = Account.UNKNOWN_ACCOUNT_ID;
      short containerId = Container.UNKNOWN_CONTAINER_ID;
      long operationTime = Utils.Infinite_Time;
      if (versionToDeserializeIn < VERSION_1 || versionToDeserializeIn > VERSION_MAX) {
        throw new IllegalArgumentException("Unknown version to deserialize MessageInfoList " + versionToDeserializeIn);
      }
      if (versionToDeserializeIn >= VERSION_5) {
        isTtlUpdated = stream.readByte() == UPDATED;
      }
      if (versionToDeserializeIn >= VERSION_6) {
        isUndeleted = stream.readByte() == UPDATED;
      }
      if (versionToDeserializeIn > VERSION_1) {
        crc = stream.readByte() == FIELD_PRESENT ? stream.readLong() : null;
      }
      if (versionToDeserializeIn > VERSION_2) {
        accountId = stream.readShort();
        containerId = stream.readShort();
        operationTime = stream.readLong();
      }
      if (versionToDeserializeIn >= VERSION_6) {
        lifeVersion = stream.readShort();
      }

      messageInfoList.add(
          new MessageInfo(id, size, isDeleted, isTtlUpdated, isUndeleted, ttl, crc, accountId, containerId,
              operationTime, lifeVersion));

      if (versionToDeserializeIn > VERSION_3) {
        MessageMetadata messageMetadata =
            stream.readByte() == FIELD_PRESENT ? MessageMetadata.deserializeMessageMetadata(stream) : null;
        messageMetadataList.add(messageMetadata);
      } else {
        messageMetadataList.add(null);
      }
    }
    return new MessageInfoAndMetadataListSerde(messageInfoList, messageMetadataList, versionToDeserializeIn);
  }
Deserialize the given stream and return the MessageInfo and Metadata lists.
@param stream the stream to deserialize from.
@param map the clustermap to use.
@param versionToDeserializeIn the SerDe version to use to deserialize.
@return the deserialized {@link MessageInfoAndMetadataListSerde}.
@throws IOException if an I/O error occurs while reading from the stream.
ambry-protocol/src/main/java/com/github/ambry/protocol/MessageInfoAndMetadataListSerde.java