@Test public void copyTest() throws Exception { storeCopier.copy(new StoreFindTokenFactory(STORE_KEY_FACTORY).getNewFindToken()); storeCopier.close(); StoreMetrics storeMetrics = new StoreMetrics(new MetricRegistry()); Files.copy(new File(srcDir, StoreDescriptor.STORE_DESCRIPTOR_FILENAME).toPath(), new File(tgtDir, StoreDescriptor.STORE_DESCRIPTOR_FILENAME).toPath(), StandardCopyOption.REPLACE_EXISTING); BlobStore tgt = new BlobStore(STORE_ID, storeConfig, null, null, DISK_IO_SCHEDULER, StoreTestUtils.DEFAULT_DISK_SPACE_ALLOCATOR, storeMetrics, storeMetrics, tgtDir.getAbsolutePath(), STORE_CAPACITY, STORE_KEY_FACTORY, null, null, time); tgt.start(); try { StoreKey[] failKeys = {expiredId, deletedId, putTtlUpdatedAndDeletedId}; for (StoreKey key : failKeys) { try { tgt.get(Collections.singletonList(key), EnumSet.allOf(StoreGetOptions.class)); fail("Should have failed to get " + key); } catch (StoreException e) { assertEquals("Unexpected StoreErrorCode", StoreErrorCodes.ID_Not_Found, e.getErrorCode()); } } Map<StoreKey, Pair<byte[], Long>> successKeys = new HashMap<>(); successKeys.put(permanentPutId, new Pair<>(permanentPutData, Utils.Infinite_Time)); successKeys.put(putAndTtlUpdatedId, new Pair<>(putAndTtlUpdatedData, Utils.Infinite_Time)); successKeys.put(temporaryPutId, new Pair<>(temporaryPutData, temporaryPutExpiryTimeMs)); for (Map.Entry<StoreKey, Pair<byte[], Long>> entry : successKeys.entrySet()) { StoreInfo storeInfo = tgt.get(Collections.singletonList(entry.getKey()), EnumSet.noneOf(StoreGetOptions.class)); MessageInfo messageInfo = storeInfo.getMessageReadSetInfo().get(0); byte[] data = entry.getValue().getFirst(); assertEquals("Size does not match", data.length, messageInfo.getSize()); assertEquals("Size does not match", data.length, storeInfo.getMessageReadSet().sizeInBytes(0)); assertFalse("Should not be deleted or expired", messageInfo.isDeleted() || messageInfo.isExpired()); assertEquals("Ttl update flag not as expected", putAndTtlUpdatedId.equals(entry.getKey()), messageInfo.isTtlUpdated()); assertEquals("Expiration time does not match", entry.getValue().getSecond().longValue(), messageInfo.getExpirationTimeInMs()); ByteBufferChannel channel = new ByteBufferChannel(ByteBuffer.allocate(data.length)); storeInfo.getMessageReadSet().writeTo(0, channel, 0, data.length); assertArrayEquals("Data put does not match data copied", data, channel.getBuffer().array()); } } finally { tgt.shutdown(); } }
public Pair<FindToken, Boolean> copy(FindToken startToken) throws Exception { boolean sourceHasProblems = false; FindToken lastToken; FindToken token = startToken; do { lastToken = token; FindInfo findInfo = src.findEntriesSince(lastToken, fetchSizeInBytes); List<MessageInfo> messageInfos = findInfo.getMessageEntries(); for (Transformer transformer : transformers) { transformer.warmup(messageInfos); } for (MessageInfo messageInfo : messageInfos) { logger.trace("Processing {} - isDeleted: {}, isExpired {}", messageInfo.getStoreKey(), messageInfo.isDeleted(), messageInfo.isExpired()); if (!messageInfo.isExpired() && !messageInfo.isDeleted()) { if (tgt.findMissingKeys(Collections.singletonList(messageInfo.getStoreKey())).size() == 1) { StoreInfo storeInfo = src.get(Collections.singletonList(messageInfo.getStoreKey()), EnumSet.allOf(StoreGetOptions.class)); MessageReadSet readSet = storeInfo.getMessageReadSet(); if (readSet.sizeInBytes(0) > Integer.MAX_VALUE) { throw new IllegalStateException("Cannot copy blobs whose size > Integer.MAX_VALUE"); } int size = (int) readSet.sizeInBytes(0); byte[] buf = new byte[size]; readSet.writeTo(0, new ByteBufferChannel(ByteBuffer.wrap(buf)), 0, size); Message message = new Message(storeInfo.getMessageReadSetInfo().get(0), new ByteArrayInputStream(buf)); for (Transformer transformer : transformers) { TransformationOutput tfmOutput = transformer.transform(message); if (tfmOutput.getException() != null) { throw tfmOutput.getException(); } else { message = tfmOutput.getMsg(); } if (message == null) { break; } } if (message == null) { logger.trace("Dropping {} because the transformers did not return a message", messageInfo.getStoreKey()); continue; } MessageFormatWriteSet writeSet = new MessageFormatWriteSet(message.getStream(), Collections.singletonList(message.getMessageInfo()), false); tgt.put(writeSet); MessageInfo tgtMsgInfo = message.getMessageInfo(); if (tgtMsgInfo.isTtlUpdated()) { MessageInfo updateMsgInfo = new MessageInfo(tgtMsgInfo.getStoreKey(), 0, false, true, tgtMsgInfo.getExpirationTimeInMs(), tgtMsgInfo.getAccountId(), tgtMsgInfo.getContainerId(), tgtMsgInfo.getOperationTimeMs()); tgt.updateTtl(Collections.singletonList(updateMsgInfo)); } logger.trace("Copied {} as {}", messageInfo.getStoreKey(), tgtMsgInfo.getStoreKey()); } else if (!messageInfo.isTtlUpdated()) { logger.warn("Found a duplicate entry for {} while copying data", messageInfo.getStoreKey()); sourceHasProblems = true; } } } token = findInfo.getFindToken(); double percentBytesRead = src.isEmpty() ? 100.0 : token.getBytesRead() * 100.0 / src.getSizeInBytes(); logger.info("[{}] [{}] {}% copied", Thread.currentThread().getName(), storeId, df.format(percentBytesRead)); } while (!token.equals(lastToken)); return new Pair<>(token, sourceHasProblems); }
StoreCopier implements Closeable { public Pair<FindToken, Boolean> copy(FindToken startToken) throws Exception { boolean sourceHasProblems = false; FindToken lastToken; FindToken token = startToken; do { lastToken = token; FindInfo findInfo = src.findEntriesSince(lastToken, fetchSizeInBytes); List<MessageInfo> messageInfos = findInfo.getMessageEntries(); for (Transformer transformer : transformers) { transformer.warmup(messageInfos); } for (MessageInfo messageInfo : messageInfos) { logger.trace("Processing {} - isDeleted: {}, isExpired {}", messageInfo.getStoreKey(), messageInfo.isDeleted(), messageInfo.isExpired()); if (!messageInfo.isExpired() && !messageInfo.isDeleted()) { if (tgt.findMissingKeys(Collections.singletonList(messageInfo.getStoreKey())).size() == 1) { StoreInfo storeInfo = src.get(Collections.singletonList(messageInfo.getStoreKey()), EnumSet.allOf(StoreGetOptions.class)); MessageReadSet readSet = storeInfo.getMessageReadSet(); if (readSet.sizeInBytes(0) > Integer.MAX_VALUE) { throw new IllegalStateException("Cannot copy blobs whose size > Integer.MAX_VALUE"); } int size = (int) readSet.sizeInBytes(0); byte[] buf = new byte[size]; readSet.writeTo(0, new ByteBufferChannel(ByteBuffer.wrap(buf)), 0, size); Message message = new Message(storeInfo.getMessageReadSetInfo().get(0), new ByteArrayInputStream(buf)); for (Transformer transformer : transformers) { TransformationOutput tfmOutput = transformer.transform(message); if (tfmOutput.getException() != null) { throw tfmOutput.getException(); } else { message = tfmOutput.getMsg(); } if (message == null) { break; } } if (message == null) { logger.trace("Dropping {} because the transformers did not return a message", messageInfo.getStoreKey()); continue; } MessageFormatWriteSet writeSet = new MessageFormatWriteSet(message.getStream(), Collections.singletonList(message.getMessageInfo()), false); tgt.put(writeSet); MessageInfo tgtMsgInfo = message.getMessageInfo(); if (tgtMsgInfo.isTtlUpdated()) { MessageInfo updateMsgInfo = new MessageInfo(tgtMsgInfo.getStoreKey(), 0, false, true, tgtMsgInfo.getExpirationTimeInMs(), tgtMsgInfo.getAccountId(), tgtMsgInfo.getContainerId(), tgtMsgInfo.getOperationTimeMs()); tgt.updateTtl(Collections.singletonList(updateMsgInfo)); } logger.trace("Copied {} as {}", messageInfo.getStoreKey(), tgtMsgInfo.getStoreKey()); } else if (!messageInfo.isTtlUpdated()) { logger.warn("Found a duplicate entry for {} while copying data", messageInfo.getStoreKey()); sourceHasProblems = true; } } } token = findInfo.getFindToken(); double percentBytesRead = src.isEmpty() ? 100.0 : token.getBytesRead() * 100.0 / src.getSizeInBytes(); logger.info("[{}] [{}] {}% copied", Thread.currentThread().getName(), storeId, df.format(percentBytesRead)); } while (!token.equals(lastToken)); return new Pair<>(token, sourceHasProblems); } }
StoreCopier implements Closeable { public Pair<FindToken, Boolean> copy(FindToken startToken) throws Exception { boolean sourceHasProblems = false; FindToken lastToken; FindToken token = startToken; do { lastToken = token; FindInfo findInfo = src.findEntriesSince(lastToken, fetchSizeInBytes); List<MessageInfo> messageInfos = findInfo.getMessageEntries(); for (Transformer transformer : transformers) { transformer.warmup(messageInfos); } for (MessageInfo messageInfo : messageInfos) { logger.trace("Processing {} - isDeleted: {}, isExpired {}", messageInfo.getStoreKey(), messageInfo.isDeleted(), messageInfo.isExpired()); if (!messageInfo.isExpired() && !messageInfo.isDeleted()) { if (tgt.findMissingKeys(Collections.singletonList(messageInfo.getStoreKey())).size() == 1) { StoreInfo storeInfo = src.get(Collections.singletonList(messageInfo.getStoreKey()), EnumSet.allOf(StoreGetOptions.class)); MessageReadSet readSet = storeInfo.getMessageReadSet(); if (readSet.sizeInBytes(0) > Integer.MAX_VALUE) { throw new IllegalStateException("Cannot copy blobs whose size > Integer.MAX_VALUE"); } int size = (int) readSet.sizeInBytes(0); byte[] buf = new byte[size]; readSet.writeTo(0, new ByteBufferChannel(ByteBuffer.wrap(buf)), 0, size); Message message = new Message(storeInfo.getMessageReadSetInfo().get(0), new ByteArrayInputStream(buf)); for (Transformer transformer : transformers) { TransformationOutput tfmOutput = transformer.transform(message); if (tfmOutput.getException() != null) { throw tfmOutput.getException(); } else { message = tfmOutput.getMsg(); } if (message == null) { break; } } if (message == null) { logger.trace("Dropping {} because the transformers did not return a message", messageInfo.getStoreKey()); continue; } MessageFormatWriteSet writeSet = new MessageFormatWriteSet(message.getStream(), Collections.singletonList(message.getMessageInfo()), false); tgt.put(writeSet); MessageInfo tgtMsgInfo = message.getMessageInfo(); if (tgtMsgInfo.isTtlUpdated()) { MessageInfo updateMsgInfo = new MessageInfo(tgtMsgInfo.getStoreKey(), 0, false, true, tgtMsgInfo.getExpirationTimeInMs(), tgtMsgInfo.getAccountId(), tgtMsgInfo.getContainerId(), tgtMsgInfo.getOperationTimeMs()); tgt.updateTtl(Collections.singletonList(updateMsgInfo)); } logger.trace("Copied {} as {}", messageInfo.getStoreKey(), tgtMsgInfo.getStoreKey()); } else if (!messageInfo.isTtlUpdated()) { logger.warn("Found a duplicate entry for {} while copying data", messageInfo.getStoreKey()); sourceHasProblems = true; } } } token = findInfo.getFindToken(); double percentBytesRead = src.isEmpty() ? 100.0 : token.getBytesRead() * 100.0 / src.getSizeInBytes(); logger.info("[{}] [{}] {}% copied", Thread.currentThread().getName(), storeId, df.format(percentBytesRead)); } while (!token.equals(lastToken)); return new Pair<>(token, sourceHasProblems); } StoreCopier(String storeId, File srcDir, File tgtDir, long storeCapacity, long fetchSizeInBytes,
      StoreConfig storeConfig, StoreMetrics metrics, StoreKeyFactory storeKeyFactory, DiskIOScheduler diskIOScheduler,
      DiskSpaceAllocator diskSpaceAllocator, List<Transformer> transformers, Time time); }
StoreCopier implements Closeable { public Pair<FindToken, Boolean> copy(FindToken startToken) throws Exception { boolean sourceHasProblems = false; FindToken lastToken; FindToken token = startToken; do { lastToken = token; FindInfo findInfo = src.findEntriesSince(lastToken, fetchSizeInBytes); List<MessageInfo> messageInfos = findInfo.getMessageEntries(); for (Transformer transformer : transformers) { transformer.warmup(messageInfos); } for (MessageInfo messageInfo : messageInfos) { logger.trace("Processing {} - isDeleted: {}, isExpired {}", messageInfo.getStoreKey(), messageInfo.isDeleted(), messageInfo.isExpired()); if (!messageInfo.isExpired() && !messageInfo.isDeleted()) { if (tgt.findMissingKeys(Collections.singletonList(messageInfo.getStoreKey())).size() == 1) { StoreInfo storeInfo = src.get(Collections.singletonList(messageInfo.getStoreKey()), EnumSet.allOf(StoreGetOptions.class)); MessageReadSet readSet = storeInfo.getMessageReadSet(); if (readSet.sizeInBytes(0) > Integer.MAX_VALUE) { throw new IllegalStateException("Cannot copy blobs whose size > Integer.MAX_VALUE"); } int size = (int) readSet.sizeInBytes(0); byte[] buf = new byte[size]; readSet.writeTo(0, new ByteBufferChannel(ByteBuffer.wrap(buf)), 0, size); Message message = new Message(storeInfo.getMessageReadSetInfo().get(0), new ByteArrayInputStream(buf)); for (Transformer transformer : transformers) { TransformationOutput tfmOutput = transformer.transform(message); if (tfmOutput.getException() != null) { throw tfmOutput.getException(); } else { message = tfmOutput.getMsg(); } if (message == null) { break; } } if (message == null) { logger.trace("Dropping {} because the transformers did not return a message", messageInfo.getStoreKey()); continue; } MessageFormatWriteSet writeSet = new MessageFormatWriteSet(message.getStream(), Collections.singletonList(message.getMessageInfo()), false); tgt.put(writeSet); MessageInfo tgtMsgInfo = message.getMessageInfo(); if (tgtMsgInfo.isTtlUpdated()) { MessageInfo updateMsgInfo = new MessageInfo(tgtMsgInfo.getStoreKey(), 0, false, true, tgtMsgInfo.getExpirationTimeInMs(), tgtMsgInfo.getAccountId(), tgtMsgInfo.getContainerId(), tgtMsgInfo.getOperationTimeMs()); tgt.updateTtl(Collections.singletonList(updateMsgInfo)); } logger.trace("Copied {} as {}", messageInfo.getStoreKey(), tgtMsgInfo.getStoreKey()); } else if (!messageInfo.isTtlUpdated()) { logger.warn("Found a duplicate entry for {} while copying data", messageInfo.getStoreKey()); sourceHasProblems = true; } } } token = findInfo.getFindToken(); double percentBytesRead = src.isEmpty() ? 100.0 : token.getBytesRead() * 100.0 / src.getSizeInBytes(); logger.info("[{}] [{}] {}% copied", Thread.currentThread().getName(), storeId, df.format(percentBytesRead)); } while (!token.equals(lastToken)); return new Pair<>(token, sourceHasProblems); } StoreCopier(String storeId, File srcDir, File tgtDir, long storeCapacity, long fetchSizeInBytes,
      StoreConfig storeConfig, StoreMetrics metrics, StoreKeyFactory storeKeyFactory, DiskIOScheduler diskIOScheduler,
      DiskSpaceAllocator diskSpaceAllocator, List<Transformer> transformers, Time time); static void main(String[] args); @Override void close(); Pair<FindToken, Boolean> copy(FindToken startToken); }
StoreCopier implements Closeable { public Pair<FindToken, Boolean> copy(FindToken startToken) throws Exception { boolean sourceHasProblems = false; FindToken lastToken; FindToken token = startToken; do { lastToken = token; FindInfo findInfo = src.findEntriesSince(lastToken, fetchSizeInBytes); List<MessageInfo> messageInfos = findInfo.getMessageEntries(); for (Transformer transformer : transformers) { transformer.warmup(messageInfos); } for (MessageInfo messageInfo : messageInfos) { logger.trace("Processing {} - isDeleted: {}, isExpired {}", messageInfo.getStoreKey(), messageInfo.isDeleted(), messageInfo.isExpired()); if (!messageInfo.isExpired() && !messageInfo.isDeleted()) { if (tgt.findMissingKeys(Collections.singletonList(messageInfo.getStoreKey())).size() == 1) { StoreInfo storeInfo = src.get(Collections.singletonList(messageInfo.getStoreKey()), EnumSet.allOf(StoreGetOptions.class)); MessageReadSet readSet = storeInfo.getMessageReadSet(); if (readSet.sizeInBytes(0) > Integer.MAX_VALUE) { throw new IllegalStateException("Cannot copy blobs whose size > Integer.MAX_VALUE"); } int size = (int) readSet.sizeInBytes(0); byte[] buf = new byte[size]; readSet.writeTo(0, new ByteBufferChannel(ByteBuffer.wrap(buf)), 0, size); Message message = new Message(storeInfo.getMessageReadSetInfo().get(0), new ByteArrayInputStream(buf)); for (Transformer transformer : transformers) { TransformationOutput tfmOutput = transformer.transform(message); if (tfmOutput.getException() != null) { throw tfmOutput.getException(); } else { message = tfmOutput.getMsg(); } if (message == null) { break; } } if (message == null) { logger.trace("Dropping {} because the transformers did not return a message", messageInfo.getStoreKey()); continue; } MessageFormatWriteSet writeSet = new MessageFormatWriteSet(message.getStream(), Collections.singletonList(message.getMessageInfo()), false); tgt.put(writeSet); MessageInfo tgtMsgInfo = message.getMessageInfo(); if (tgtMsgInfo.isTtlUpdated()) { MessageInfo updateMsgInfo = new MessageInfo(tgtMsgInfo.getStoreKey(), 0, false, true, tgtMsgInfo.getExpirationTimeInMs(), tgtMsgInfo.getAccountId(), tgtMsgInfo.getContainerId(), tgtMsgInfo.getOperationTimeMs()); tgt.updateTtl(Collections.singletonList(updateMsgInfo)); } logger.trace("Copied {} as {}", messageInfo.getStoreKey(), tgtMsgInfo.getStoreKey()); } else if (!messageInfo.isTtlUpdated()) { logger.warn("Found a duplicate entry for {} while copying data", messageInfo.getStoreKey()); sourceHasProblems = true; } } } token = findInfo.getFindToken(); double percentBytesRead = src.isEmpty() ? 100.0 : token.getBytesRead() * 100.0 / src.getSizeInBytes(); logger.info("[{}] [{}] {}% copied", Thread.currentThread().getName(), storeId, df.format(percentBytesRead)); } while (!token.equals(lastToken)); return new Pair<>(token, sourceHasProblems); } StoreCopier(String storeId, File srcDir, File tgtDir, long storeCapacity, long fetchSizeInBytes,
      StoreConfig storeConfig, StoreMetrics metrics, StoreKeyFactory storeKeyFactory, DiskIOScheduler diskIOScheduler,
      DiskSpaceAllocator diskSpaceAllocator, List<Transformer> transformers, Time time); static void main(String[] args); @Override void close(); Pair<FindToken, Boolean> copy(FindToken startToken);  }