@Test public void testEndProcessingTime() { Collection<String> idsToDelete = createTestIds(50); doReturn(createBatchWriteItemResult()).when(mockAmazonDynamoDB).batchWriteItem(any(BatchWriteItemRequest.class)); ConsumerResult<String> consumerResult = batchedDeleter.consume(idsToDelete, Instant.now().minusSeconds(60)); Assert.assertEquals(consumerResult.getItemsConsumedCount(), BatchedDeleter.MAX_BATCH_WRITE_ITEM); verify(mockAmazonDynamoDB, times(1)).batchWriteItem(any(BatchWriteItemRequest.class)); }
@Override public ConsumerResult<String> consume(Collection<String> collection, Instant endProcessingTime) { if(collection == null) { logger.info("Input collection is null. No deletes to perform on {}", tableName); return new ConsumerResult<String>().setItemsConsumedCount(0); } logger.info("Attempting to delete {} items from {}", collection.size(), tableName); Collection<List<String>> batchedSets = CollectionUtil.split(new LinkedList<>(collection), MAX_BATCH_WRITE_ITEM); int itemsRemoved = 0; for(List<String> deleteBatch : batchedSets) { try { itemsRemoved += deleteItems(deleteBatch); } catch(Exception e) { logger.error(String.format("Delete operation failed on %1$s. Interrupting processing.", tableName), e); return new ConsumerResult<String>().setInterrupted(true); } if(InstantUtil.isNowAfterOrEqual(endProcessingTime)) break; if(!delay(deleteCallDelayMillis)) break; } logger.info("Deleted {} items from {}", tableName, itemsRemoved); return new ConsumerResult<String>().setItemsConsumedCount(itemsRemoved); }
BatchedDeleter extends BaseBatchedOperation implements Consumer<String> { @Override public ConsumerResult<String> consume(Collection<String> collection, Instant endProcessingTime) { if(collection == null) { logger.info("Input collection is null. No deletes to perform on {}", tableName); return new ConsumerResult<String>().setItemsConsumedCount(0); } logger.info("Attempting to delete {} items from {}", collection.size(), tableName); Collection<List<String>> batchedSets = CollectionUtil.split(new LinkedList<>(collection), MAX_BATCH_WRITE_ITEM); int itemsRemoved = 0; for(List<String> deleteBatch : batchedSets) { try { itemsRemoved += deleteItems(deleteBatch); } catch(Exception e) { logger.error(String.format("Delete operation failed on %1$s. Interrupting processing.", tableName), e); return new ConsumerResult<String>().setInterrupted(true); } if(InstantUtil.isNowAfterOrEqual(endProcessingTime)) break; if(!delay(deleteCallDelayMillis)) break; } logger.info("Deleted {} items from {}", tableName, itemsRemoved); return new ConsumerResult<String>().setItemsConsumedCount(itemsRemoved); } }
BatchedDeleter extends BaseBatchedOperation implements Consumer<String> { @Override public ConsumerResult<String> consume(Collection<String> collection, Instant endProcessingTime) { if(collection == null) { logger.info("Input collection is null. No deletes to perform on {}", tableName); return new ConsumerResult<String>().setItemsConsumedCount(0); } logger.info("Attempting to delete {} items from {}", collection.size(), tableName); Collection<List<String>> batchedSets = CollectionUtil.split(new LinkedList<>(collection), MAX_BATCH_WRITE_ITEM); int itemsRemoved = 0; for(List<String> deleteBatch : batchedSets) { try { itemsRemoved += deleteItems(deleteBatch); } catch(Exception e) { logger.error(String.format("Delete operation failed on %1$s. Interrupting processing.", tableName), e); return new ConsumerResult<String>().setInterrupted(true); } if(InstantUtil.isNowAfterOrEqual(endProcessingTime)) break; if(!delay(deleteCallDelayMillis)) break; } logger.info("Deleted {} items from {}", tableName, itemsRemoved); return new ConsumerResult<String>().setItemsConsumedCount(itemsRemoved); } BatchedDeleter(AmazonDynamoDB dynamoDB,
        String tableName,
        String fieldName); }
BatchedDeleter extends BaseBatchedOperation implements Consumer<String> { @Override public ConsumerResult<String> consume(Collection<String> collection, Instant endProcessingTime) { if(collection == null) { logger.info("Input collection is null. No deletes to perform on {}", tableName); return new ConsumerResult<String>().setItemsConsumedCount(0); } logger.info("Attempting to delete {} items from {}", collection.size(), tableName); Collection<List<String>> batchedSets = CollectionUtil.split(new LinkedList<>(collection), MAX_BATCH_WRITE_ITEM); int itemsRemoved = 0; for(List<String> deleteBatch : batchedSets) { try { itemsRemoved += deleteItems(deleteBatch); } catch(Exception e) { logger.error(String.format("Delete operation failed on %1$s. Interrupting processing.", tableName), e); return new ConsumerResult<String>().setInterrupted(true); } if(InstantUtil.isNowAfterOrEqual(endProcessingTime)) break; if(!delay(deleteCallDelayMillis)) break; } logger.info("Deleted {} items from {}", tableName, itemsRemoved); return new ConsumerResult<String>().setItemsConsumedCount(itemsRemoved); } BatchedDeleter(AmazonDynamoDB dynamoDB,
        String tableName,
        String fieldName); @Override ConsumerResult<String> consume(Collection<String> collection, Instant endProcessingTime); BatchedDeleter setDeleteCallDelayMillis(long deleteCallDelayMillis); BatchedDeleter setLogDeleteOnly(boolean logDeleteOnly); @Override String toString(); }
BatchedDeleter extends BaseBatchedOperation implements Consumer<String> { @Override public ConsumerResult<String> consume(Collection<String> collection, Instant endProcessingTime) { if(collection == null) { logger.info("Input collection is null. No deletes to perform on {}", tableName); return new ConsumerResult<String>().setItemsConsumedCount(0); } logger.info("Attempting to delete {} items from {}", collection.size(), tableName); Collection<List<String>> batchedSets = CollectionUtil.split(new LinkedList<>(collection), MAX_BATCH_WRITE_ITEM); int itemsRemoved = 0; for(List<String> deleteBatch : batchedSets) { try { itemsRemoved += deleteItems(deleteBatch); } catch(Exception e) { logger.error(String.format("Delete operation failed on %1$s. Interrupting processing.", tableName), e); return new ConsumerResult<String>().setInterrupted(true); } if(InstantUtil.isNowAfterOrEqual(endProcessingTime)) break; if(!delay(deleteCallDelayMillis)) break; } logger.info("Deleted {} items from {}", tableName, itemsRemoved); return new ConsumerResult<String>().setItemsConsumedCount(itemsRemoved); } BatchedDeleter(AmazonDynamoDB dynamoDB,
        String tableName,
        String fieldName); @Override ConsumerResult<String> consume(Collection<String> collection, Instant endProcessingTime); BatchedDeleter setDeleteCallDelayMillis(long deleteCallDelayMillis); BatchedDeleter setLogDeleteOnly(boolean logDeleteOnly); @Override String toString(); final static int MAX_BATCH_WRITE_ITEM; }