@Test public void recoveryTest() throws MessageFormatException, IOException { MessageStoreRecovery recovery = new BlobStoreRecovery(); ReadImp readrecovery = new ReadImp(); readrecovery.initialize(); List<MessageInfo> recoveredMessages = recovery.recover(readrecovery, 0, readrecovery.getSize(), new MockIdFactory()); Assert.assertEquals(recoveredMessages.size(), 6); verifyInfo(recoveredMessages.get(0), readrecovery.keys[0], readrecovery.sizes.get(0), readrecovery.expectedExpirationTimeMs, false, false); verifyInfo(recoveredMessages.get(1), readrecovery.keys[1], readrecovery.sizes.get(1), Utils.Infinite_Time, false, false); verifyInfo(recoveredMessages.get(2), readrecovery.keys[2], readrecovery.sizes.get(2), Utils.Infinite_Time, false, false); verifyInfo(recoveredMessages.get(4), readrecovery.keys[1], readrecovery.sizes.get(4), Utils.Infinite_Time, true, false); if (MessageFormatRecord.headerVersionToUse >= MessageFormatRecord.Message_Header_Version_V2) { verifyInfo(recoveredMessages.get(3), readrecovery.keys[1], readrecovery.sizes.get(3), Utils.Infinite_Time, false, true); verifyInfo(recoveredMessages.get(5), readrecovery.keys[0], readrecovery.sizes.get(5), Utils.Infinite_Time, false, true); } else { verifyInfo(recoveredMessages.get(3), readrecovery.keys[3], readrecovery.sizes.get(3), Utils.Infinite_Time, false, false); verifyInfo(recoveredMessages.get(5), readrecovery.keys[4], readrecovery.sizes.get(5), Utils.Infinite_Time, false, false); } }
@Override public List<MessageInfo> recover(Read read, long startOffset, long endOffset, StoreKeyFactory factory) throws IOException { ArrayList<MessageInfo> messageRecovered = new ArrayList<MessageInfo>(); try { while (startOffset < endOffset) { ByteBuffer headerVersion = ByteBuffer.allocate(Version_Field_Size_In_Bytes); if (startOffset + Version_Field_Size_In_Bytes > endOffset) { throw new IndexOutOfBoundsException("Unable to read version. Reached end of stream"); } read.readInto(headerVersion, startOffset); startOffset += headerVersion.capacity(); headerVersion.flip(); short version = headerVersion.getShort(); if (!isValidHeaderVersion(version)) { throw new MessageFormatException("Version not known while reading message - " + version, MessageFormatErrorCodes.Unknown_Format_Version); } ByteBuffer header = ByteBuffer.allocate(getHeaderSizeForVersion(version)); header.putShort(version); if (startOffset + (header.capacity() - headerVersion.capacity()) > endOffset) { throw new IndexOutOfBoundsException("Unable to read version. Reached end of stream"); } read.readInto(header, startOffset); startOffset += header.capacity() - headerVersion.capacity(); header.flip(); MessageHeader_Format headerFormat = getMessageHeader(version, header); headerFormat.verifyHeader(); ReadInputStream stream = new ReadInputStream(read, startOffset, endOffset); StoreKey key = factory.getStoreKey(new DataInputStream(stream)); short lifeVersion = 0; if (headerFormat.hasLifeVersion()) { lifeVersion = headerFormat.getLifeVersion(); } if (headerFormat.isPutRecord()) { 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) { logger.error("Message format exception while recovering messages", e); } catch (IndexOutOfBoundsException e) { logger.error("Trying to read more than the available bytes"); } for (MessageInfo messageInfo : messageRecovered) { logger.info("Message Recovered key {} size {} ttl {} deleted {} undelete {}", messageInfo.getStoreKey(), messageInfo.getSize(), messageInfo.getExpirationTimeInMs(), messageInfo.isDeleted(), messageInfo.isUndeleted()); } return messageRecovered; }
BlobStoreRecovery implements MessageStoreRecovery { @Override public List<MessageInfo> recover(Read read, long startOffset, long endOffset, StoreKeyFactory factory) throws IOException { ArrayList<MessageInfo> messageRecovered = new ArrayList<MessageInfo>(); try { while (startOffset < endOffset) { ByteBuffer headerVersion = ByteBuffer.allocate(Version_Field_Size_In_Bytes); if (startOffset + Version_Field_Size_In_Bytes > endOffset) { throw new IndexOutOfBoundsException("Unable to read version. Reached end of stream"); } read.readInto(headerVersion, startOffset); startOffset += headerVersion.capacity(); headerVersion.flip(); short version = headerVersion.getShort(); if (!isValidHeaderVersion(version)) { throw new MessageFormatException("Version not known while reading message - " + version, MessageFormatErrorCodes.Unknown_Format_Version); } ByteBuffer header = ByteBuffer.allocate(getHeaderSizeForVersion(version)); header.putShort(version); if (startOffset + (header.capacity() - headerVersion.capacity()) > endOffset) { throw new IndexOutOfBoundsException("Unable to read version. Reached end of stream"); } read.readInto(header, startOffset); startOffset += header.capacity() - headerVersion.capacity(); header.flip(); MessageHeader_Format headerFormat = getMessageHeader(version, header); headerFormat.verifyHeader(); ReadInputStream stream = new ReadInputStream(read, startOffset, endOffset); StoreKey key = factory.getStoreKey(new DataInputStream(stream)); short lifeVersion = 0; if (headerFormat.hasLifeVersion()) { lifeVersion = headerFormat.getLifeVersion(); } if (headerFormat.isPutRecord()) { 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) { logger.error("Message format exception while recovering messages", e); } catch (IndexOutOfBoundsException e) { logger.error("Trying to read more than the available bytes"); } for (MessageInfo messageInfo : messageRecovered) { logger.info("Message Recovered key {} size {} ttl {} deleted {} undelete {}", messageInfo.getStoreKey(), messageInfo.getSize(), messageInfo.getExpirationTimeInMs(), messageInfo.isDeleted(), messageInfo.isUndeleted()); } return messageRecovered; } }
BlobStoreRecovery implements MessageStoreRecovery { @Override public List<MessageInfo> recover(Read read, long startOffset, long endOffset, StoreKeyFactory factory) throws IOException { ArrayList<MessageInfo> messageRecovered = new ArrayList<MessageInfo>(); try { while (startOffset < endOffset) { ByteBuffer headerVersion = ByteBuffer.allocate(Version_Field_Size_In_Bytes); if (startOffset + Version_Field_Size_In_Bytes > endOffset) { throw new IndexOutOfBoundsException("Unable to read version. Reached end of stream"); } read.readInto(headerVersion, startOffset); startOffset += headerVersion.capacity(); headerVersion.flip(); short version = headerVersion.getShort(); if (!isValidHeaderVersion(version)) { throw new MessageFormatException("Version not known while reading message - " + version, MessageFormatErrorCodes.Unknown_Format_Version); } ByteBuffer header = ByteBuffer.allocate(getHeaderSizeForVersion(version)); header.putShort(version); if (startOffset + (header.capacity() - headerVersion.capacity()) > endOffset) { throw new IndexOutOfBoundsException("Unable to read version. Reached end of stream"); } read.readInto(header, startOffset); startOffset += header.capacity() - headerVersion.capacity(); header.flip(); MessageHeader_Format headerFormat = getMessageHeader(version, header); headerFormat.verifyHeader(); ReadInputStream stream = new ReadInputStream(read, startOffset, endOffset); StoreKey key = factory.getStoreKey(new DataInputStream(stream)); short lifeVersion = 0; if (headerFormat.hasLifeVersion()) { lifeVersion = headerFormat.getLifeVersion(); } if (headerFormat.isPutRecord()) { 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) { logger.error("Message format exception while recovering messages", e); } catch (IndexOutOfBoundsException e) { logger.error("Trying to read more than the available bytes"); } for (MessageInfo messageInfo : messageRecovered) { logger.info("Message Recovered key {} size {} ttl {} deleted {} undelete {}", messageInfo.getStoreKey(), messageInfo.getSize(), messageInfo.getExpirationTimeInMs(), messageInfo.isDeleted(), messageInfo.isUndeleted()); } return messageRecovered; }  }
BlobStoreRecovery implements MessageStoreRecovery { @Override public List<MessageInfo> recover(Read read, long startOffset, long endOffset, StoreKeyFactory factory) throws IOException { ArrayList<MessageInfo> messageRecovered = new ArrayList<MessageInfo>(); try { while (startOffset < endOffset) { ByteBuffer headerVersion = ByteBuffer.allocate(Version_Field_Size_In_Bytes); if (startOffset + Version_Field_Size_In_Bytes > endOffset) { throw new IndexOutOfBoundsException("Unable to read version. Reached end of stream"); } read.readInto(headerVersion, startOffset); startOffset += headerVersion.capacity(); headerVersion.flip(); short version = headerVersion.getShort(); if (!isValidHeaderVersion(version)) { throw new MessageFormatException("Version not known while reading message - " + version, MessageFormatErrorCodes.Unknown_Format_Version); } ByteBuffer header = ByteBuffer.allocate(getHeaderSizeForVersion(version)); header.putShort(version); if (startOffset + (header.capacity() - headerVersion.capacity()) > endOffset) { throw new IndexOutOfBoundsException("Unable to read version. Reached end of stream"); } read.readInto(header, startOffset); startOffset += header.capacity() - headerVersion.capacity(); header.flip(); MessageHeader_Format headerFormat = getMessageHeader(version, header); headerFormat.verifyHeader(); ReadInputStream stream = new ReadInputStream(read, startOffset, endOffset); StoreKey key = factory.getStoreKey(new DataInputStream(stream)); short lifeVersion = 0; if (headerFormat.hasLifeVersion()) { lifeVersion = headerFormat.getLifeVersion(); } if (headerFormat.isPutRecord()) { 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) { logger.error("Message format exception while recovering messages", e); } catch (IndexOutOfBoundsException e) { logger.error("Trying to read more than the available bytes"); } for (MessageInfo messageInfo : messageRecovered) { logger.info("Message Recovered key {} size {} ttl {} deleted {} undelete {}", messageInfo.getStoreKey(), messageInfo.getSize(), messageInfo.getExpirationTimeInMs(), messageInfo.isDeleted(), messageInfo.isUndeleted()); } return messageRecovered; }  @Override List<MessageInfo> recover(Read read, long startOffset, long endOffset, StoreKeyFactory factory); }
BlobStoreRecovery implements MessageStoreRecovery { @Override public List<MessageInfo> recover(Read read, long startOffset, long endOffset, StoreKeyFactory factory) throws IOException { ArrayList<MessageInfo> messageRecovered = new ArrayList<MessageInfo>(); try { while (startOffset < endOffset) { ByteBuffer headerVersion = ByteBuffer.allocate(Version_Field_Size_In_Bytes); if (startOffset + Version_Field_Size_In_Bytes > endOffset) { throw new IndexOutOfBoundsException("Unable to read version. Reached end of stream"); } read.readInto(headerVersion, startOffset); startOffset += headerVersion.capacity(); headerVersion.flip(); short version = headerVersion.getShort(); if (!isValidHeaderVersion(version)) { throw new MessageFormatException("Version not known while reading message - " + version, MessageFormatErrorCodes.Unknown_Format_Version); } ByteBuffer header = ByteBuffer.allocate(getHeaderSizeForVersion(version)); header.putShort(version); if (startOffset + (header.capacity() - headerVersion.capacity()) > endOffset) { throw new IndexOutOfBoundsException("Unable to read version. Reached end of stream"); } read.readInto(header, startOffset); startOffset += header.capacity() - headerVersion.capacity(); header.flip(); MessageHeader_Format headerFormat = getMessageHeader(version, header); headerFormat.verifyHeader(); ReadInputStream stream = new ReadInputStream(read, startOffset, endOffset); StoreKey key = factory.getStoreKey(new DataInputStream(stream)); short lifeVersion = 0; if (headerFormat.hasLifeVersion()) { lifeVersion = headerFormat.getLifeVersion(); } if (headerFormat.isPutRecord()) { 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) { logger.error("Message format exception while recovering messages", e); } catch (IndexOutOfBoundsException e) { logger.error("Trying to read more than the available bytes"); } for (MessageInfo messageInfo : messageRecovered) { logger.info("Message Recovered key {} size {} ttl {} deleted {} undelete {}", messageInfo.getStoreKey(), messageInfo.getSize(), messageInfo.getExpirationTimeInMs(), messageInfo.isDeleted(), messageInfo.isUndeleted()); } return messageRecovered; }  @Override List<MessageInfo> recover(Read read, long startOffset, long endOffset, StoreKeyFactory factory);  }