@Override
  public RecoveryResult recover(Read read, long startOffset, long endOffset, StoreKeyFactory factory) {
    ArrayList<MessageInfo> messageRecovered = new ArrayList<MessageInfo>();
    StoreException recoveryException = null;
    try {
      while (startOffset < endOffset) {
        // read message header
        long currentIndex = startOffset;
        ByteBuffer headerVersion = ByteBuffer.allocate(Version_Field_Size_In_Bytes);
        if (currentIndex + Version_Field_Size_In_Bytes > endOffset) {
          throw new IndexOutOfBoundsException("Unable to read version. Reached end of stream");
        }
        read.readInto(headerVersion, currentIndex);
        currentIndex += headerVersion.capacity();
        headerVersion.flip();
        short version = headerVersion.getShort();
        if (!isValidHeaderVersion(version)) {
          throw new MessageFormatException("Version not known while reading message - " + version,
              MessageFormatErrorCodes.UnknownFormatVersion);
        }
        ByteBuffer header = ByteBuffer.allocate(getHeaderSizeForVersion(version));
        header.putShort(version);
        if (currentIndex + (header.capacity() - headerVersion.capacity()) > endOffset) {
          throw new IndexOutOfBoundsException("Unable to read version. Reached end of stream");
        }
        read.readInto(header, currentIndex);
        currentIndex += header.capacity() - headerVersion.capacity();
        header.flip();
        MessageHeader_Format headerFormat = getMessageHeader(version, header);
        headerFormat.verifyHeader();
        ReadInputStream stream = new ReadInputStream(read, currentIndex, endOffset);
        StoreKey key = factory.getStoreKey(new DataInputStream(stream));

        short lifeVersion = 0;
        if (headerFormat.hasLifeVersion()) {
          lifeVersion = headerFormat.getLifeVersion();
        }
        // read the appropriate type of message based on the relative offset that is set
        if (headerFormat.isPutRecord()) {
          // we do not use blob encryption key, user metadata or blob during recovery but we still deserialize
          // them to check for validity
          if (headerFormat.hasEncryptionKeyRecord()) {
            deserializeBlobEncryptionKey(stream);
          }
          BlobProperties properties = deserializeBlobProperties(stream);
          deserializeUserMetadata(stream);
          deserializeBlob(stream);
          MessageInfo info =
              new MessageInfo(key, header.capacity() + key.sizeInBytes() + headerFormat.getMessageSize(), false, false,
                  false,
                  Utils.addSecondsToEpochTime(properties.getCreationTimeInMs(), properties.getTimeToLiveInSeconds()),
                  null, properties.getAccountId(), properties.getContainerId(), properties.getCreationTimeInMs(),
                  lifeVersion);
          messageRecovered.add(info);
        } else {
          UpdateRecord updateRecord = deserializeUpdateRecord(stream);
          boolean deleted = false, ttlUpdated = false, undeleted = false;
          switch (updateRecord.getType()) {
            case DELETE:
              deleted = true;
              break;
            case TTL_UPDATE:
              ttlUpdated = true;
              break;
            case UNDELETE:
              undeleted = true;
              break;
            default:
              throw new IllegalStateException("Unknown update record type: " + updateRecord.getType());
          }
          MessageInfo info =
              new MessageInfo(key, header.capacity() + key.sizeInBytes() + headerFormat.getMessageSize(), deleted,
                  ttlUpdated, undeleted, updateRecord.getAccountId(), updateRecord.getContainerId(),
                  updateRecord.getUpdateTimeInMs(), lifeVersion);
          messageRecovered.add(info);
        }
        startOffset = stream.getCurrentPosition();
      }
    } catch (MessageFormatException e) {
      // log in case where we were not able to parse a message. we stop recovery at that point and return the
      // messages that have been recovered so far.
      logger.error("Message format exception while recovering messages, startOffset is {}, endOffset is {}",
          startOffset, endOffset, e);
      recoveryException = new StoreException(e, StoreErrorCodes.LogFileFormatError);
    } catch (IndexOutOfBoundsException e) {
      // log in case where were not able to read a complete message. we stop recovery at that point and return
      // the message that have been recovered so far.
      logger.error("Trying to read more than the available bytes, startOffset is {}, endOffset is {}", startOffset,
          endOffset);
      recoveryException = new StoreException(e, StoreErrorCodes.LogFileFormatError);
    } catch (Throwable throwable) {
      logger.error("Unexpected exception, startOffset is {}, endOffset is {}", startOffset, endOffset);
      if (throwable instanceof StoreException) {
        recoveryException = (StoreException) throwable;
      } else {
        recoveryException = new StoreException(throwable, StoreErrorCodes.LogFileFormatError);
      }
    }
    for (MessageInfo messageInfo : messageRecovered) {
      logger.info("Message Recovered key {} size {} ttl {} deleted {} undelete {}", messageInfo.getStoreKey(),
          messageInfo.getSize(), messageInfo.getExpirationTimeInMs(), messageInfo.isDeleted(),
          messageInfo.isUndeleted());
    }
    return new RecoveryResult(messageRecovered, recoveryException, startOffset);
  }
Recovers a set of messages from a given start and end offset
from the read interface that represents the underlying store
ambry-messageformat/src/main/java/com/github/ambry/messageformat/BlobStoreRecovery.java