@Test public void testCompactPartitions() throws Exception { assertEquals(0, compactor.compactPartitions()); int numPartitions = 40; String defaultClass = MockClusterMap.DEFAULT_PARTITION_CLASS; for (int i = 0; i < numPartitions; i++) { partitionMap.put(new MockPartitionId(i, defaultClass), null); when(mockDest.compactPartition(eq(Integer.toString(i)))).thenReturn(pageSize); } assertEquals(pageSize * numPartitions, compactor.compactPartitions()); assertEquals(0, vcrMetrics.compactionFailureCount.getCount()); partitionMap.remove(new MockPartitionId(0, defaultClass)); assertEquals(pageSize * (numPartitions - 1), compactor.compactPartitions()); assertEquals(0, vcrMetrics.compactionFailureCount.getCount()); CloudStorageException csex = new CloudStorageException("failure", new RuntimeException("Don't hurt me!")); when(mockDest.compactPartition(eq("2"))).thenThrow(csex); when(mockDest.compactPartition(eq("20"))).thenThrow(csex); assertEquals(pageSize * (numPartitions - 3), compactor.compactPartitions()); assertEquals(2, vcrMetrics.compactionFailureCount.getCount()); assertFalse("Should not be shutting down yet", compactor.isShutDown()); compactor.shutdown(); assertTrue("Should be shutting down now", compactor.isShutDown()); }
public int compactPartitions() { if (partitions.isEmpty()) { logger.info("Skipping compaction as no partitions are assigned."); return 0; } List<PartitionId> partitionSnapshot = new ArrayList<>(partitions); long compactionStartTime = System.currentTimeMillis(); long timeToQuit = System.currentTimeMillis() + compactionTimeLimitMs; int compactionInProgress = 0; doneLatch.set(new CountDownLatch(1)); int totalBlobsPurged = 0; int compactedPartitionCount = 0; try { while (!partitionSnapshot.isEmpty()) { while (compactionInProgress < numThreads) { if (partitionSnapshot.isEmpty()) { break; } PartitionId partitionId = partitionSnapshot.remove(0); executorCompletionService.submit(() -> compactPartition(partitionId)); compactionInProgress++; } try { totalBlobsPurged += executorCompletionService.take().get(); compactedPartitionCount++; } catch (ExecutionException ex) { vcrMetrics.compactionFailureCount.inc(); } compactionInProgress--; if (System.currentTimeMillis() >= timeToQuit) { logger.info("Compaction terminated due to time limit exceeded."); break; } if (isShutDown()) { logger.info("Compaction terminated due to shut down."); break; } } while (compactionInProgress > 0) { try { totalBlobsPurged += executorCompletionService.take().get(); compactedPartitionCount++; } catch (ExecutionException ex) { vcrMetrics.compactionFailureCount.inc(); } compactionInProgress--; } doneLatch.get().countDown(); } catch (Throwable th) { logger.error("Hit exception running compaction task", th); } finally { long compactionTime = (System.currentTimeMillis() - compactionStartTime) / 1000; logger.info("Purged {} blobs in {} partitions taking {} seconds", totalBlobsPurged, compactedPartitionCount, compactionTime); } return totalBlobsPurged; }
CloudStorageCompactor implements Runnable { public int compactPartitions() { if (partitions.isEmpty()) { logger.info("Skipping compaction as no partitions are assigned."); return 0; } List<PartitionId> partitionSnapshot = new ArrayList<>(partitions); long compactionStartTime = System.currentTimeMillis(); long timeToQuit = System.currentTimeMillis() + compactionTimeLimitMs; int compactionInProgress = 0; doneLatch.set(new CountDownLatch(1)); int totalBlobsPurged = 0; int compactedPartitionCount = 0; try { while (!partitionSnapshot.isEmpty()) { while (compactionInProgress < numThreads) { if (partitionSnapshot.isEmpty()) { break; } PartitionId partitionId = partitionSnapshot.remove(0); executorCompletionService.submit(() -> compactPartition(partitionId)); compactionInProgress++; } try { totalBlobsPurged += executorCompletionService.take().get(); compactedPartitionCount++; } catch (ExecutionException ex) { vcrMetrics.compactionFailureCount.inc(); } compactionInProgress--; if (System.currentTimeMillis() >= timeToQuit) { logger.info("Compaction terminated due to time limit exceeded."); break; } if (isShutDown()) { logger.info("Compaction terminated due to shut down."); break; } } while (compactionInProgress > 0) { try { totalBlobsPurged += executorCompletionService.take().get(); compactedPartitionCount++; } catch (ExecutionException ex) { vcrMetrics.compactionFailureCount.inc(); } compactionInProgress--; } doneLatch.get().countDown(); } catch (Throwable th) { logger.error("Hit exception running compaction task", th); } finally { long compactionTime = (System.currentTimeMillis() - compactionStartTime) / 1000; logger.info("Purged {} blobs in {} partitions taking {} seconds", totalBlobsPurged, compactedPartitionCount, compactionTime); } return totalBlobsPurged; } }
CloudStorageCompactor implements Runnable { public int compactPartitions() { if (partitions.isEmpty()) { logger.info("Skipping compaction as no partitions are assigned."); return 0; } List<PartitionId> partitionSnapshot = new ArrayList<>(partitions); long compactionStartTime = System.currentTimeMillis(); long timeToQuit = System.currentTimeMillis() + compactionTimeLimitMs; int compactionInProgress = 0; doneLatch.set(new CountDownLatch(1)); int totalBlobsPurged = 0; int compactedPartitionCount = 0; try { while (!partitionSnapshot.isEmpty()) { while (compactionInProgress < numThreads) { if (partitionSnapshot.isEmpty()) { break; } PartitionId partitionId = partitionSnapshot.remove(0); executorCompletionService.submit(() -> compactPartition(partitionId)); compactionInProgress++; } try { totalBlobsPurged += executorCompletionService.take().get(); compactedPartitionCount++; } catch (ExecutionException ex) { vcrMetrics.compactionFailureCount.inc(); } compactionInProgress--; if (System.currentTimeMillis() >= timeToQuit) { logger.info("Compaction terminated due to time limit exceeded."); break; } if (isShutDown()) { logger.info("Compaction terminated due to shut down."); break; } } while (compactionInProgress > 0) { try { totalBlobsPurged += executorCompletionService.take().get(); compactedPartitionCount++; } catch (ExecutionException ex) { vcrMetrics.compactionFailureCount.inc(); } compactionInProgress--; } doneLatch.get().countDown(); } catch (Throwable th) { logger.error("Hit exception running compaction task", th); } finally { long compactionTime = (System.currentTimeMillis() - compactionStartTime) / 1000; logger.info("Purged {} blobs in {} partitions taking {} seconds", totalBlobsPurged, compactedPartitionCount, compactionTime); } return totalBlobsPurged; } CloudStorageCompactor(CloudDestination cloudDestination, CloudConfig cloudConfig, Set<PartitionId> partitions,
      VcrMetrics vcrMetrics); }
CloudStorageCompactor implements Runnable { public int compactPartitions() { if (partitions.isEmpty()) { logger.info("Skipping compaction as no partitions are assigned."); return 0; } List<PartitionId> partitionSnapshot = new ArrayList<>(partitions); long compactionStartTime = System.currentTimeMillis(); long timeToQuit = System.currentTimeMillis() + compactionTimeLimitMs; int compactionInProgress = 0; doneLatch.set(new CountDownLatch(1)); int totalBlobsPurged = 0; int compactedPartitionCount = 0; try { while (!partitionSnapshot.isEmpty()) { while (compactionInProgress < numThreads) { if (partitionSnapshot.isEmpty()) { break; } PartitionId partitionId = partitionSnapshot.remove(0); executorCompletionService.submit(() -> compactPartition(partitionId)); compactionInProgress++; } try { totalBlobsPurged += executorCompletionService.take().get(); compactedPartitionCount++; } catch (ExecutionException ex) { vcrMetrics.compactionFailureCount.inc(); } compactionInProgress--; if (System.currentTimeMillis() >= timeToQuit) { logger.info("Compaction terminated due to time limit exceeded."); break; } if (isShutDown()) { logger.info("Compaction terminated due to shut down."); break; } } while (compactionInProgress > 0) { try { totalBlobsPurged += executorCompletionService.take().get(); compactedPartitionCount++; } catch (ExecutionException ex) { vcrMetrics.compactionFailureCount.inc(); } compactionInProgress--; } doneLatch.get().countDown(); } catch (Throwable th) { logger.error("Hit exception running compaction task", th); } finally { long compactionTime = (System.currentTimeMillis() - compactionStartTime) / 1000; logger.info("Purged {} blobs in {} partitions taking {} seconds", totalBlobsPurged, compactedPartitionCount, compactionTime); } return totalBlobsPurged; } CloudStorageCompactor(CloudDestination cloudDestination, CloudConfig cloudConfig, Set<PartitionId> partitions,
      VcrMetrics vcrMetrics); @Override void run(); void shutdown(); int compactPartitions(); }
CloudStorageCompactor implements Runnable { public int compactPartitions() { if (partitions.isEmpty()) { logger.info("Skipping compaction as no partitions are assigned."); return 0; } List<PartitionId> partitionSnapshot = new ArrayList<>(partitions); long compactionStartTime = System.currentTimeMillis(); long timeToQuit = System.currentTimeMillis() + compactionTimeLimitMs; int compactionInProgress = 0; doneLatch.set(new CountDownLatch(1)); int totalBlobsPurged = 0; int compactedPartitionCount = 0; try { while (!partitionSnapshot.isEmpty()) { while (compactionInProgress < numThreads) { if (partitionSnapshot.isEmpty()) { break; } PartitionId partitionId = partitionSnapshot.remove(0); executorCompletionService.submit(() -> compactPartition(partitionId)); compactionInProgress++; } try { totalBlobsPurged += executorCompletionService.take().get(); compactedPartitionCount++; } catch (ExecutionException ex) { vcrMetrics.compactionFailureCount.inc(); } compactionInProgress--; if (System.currentTimeMillis() >= timeToQuit) { logger.info("Compaction terminated due to time limit exceeded."); break; } if (isShutDown()) { logger.info("Compaction terminated due to shut down."); break; } } while (compactionInProgress > 0) { try { totalBlobsPurged += executorCompletionService.take().get(); compactedPartitionCount++; } catch (ExecutionException ex) { vcrMetrics.compactionFailureCount.inc(); } compactionInProgress--; } doneLatch.get().countDown(); } catch (Throwable th) { logger.error("Hit exception running compaction task", th); } finally { long compactionTime = (System.currentTimeMillis() - compactionStartTime) / 1000; logger.info("Purged {} blobs in {} partitions taking {} seconds", totalBlobsPurged, compactedPartitionCount, compactionTime); } return totalBlobsPurged; } CloudStorageCompactor(CloudDestination cloudDestination, CloudConfig cloudConfig, Set<PartitionId> partitions,
      VcrMetrics vcrMetrics); @Override void run(); void shutdown(); int compactPartitions();  }