@Test public void testCompactionStopsAfterPurgeLimit() throws Exception { int purgeLimit = numBlobsPerQuery * 2; configProps.setProperty(CloudConfig.CLOUD_COMPACTION_PURGE_LIMIT, String.valueOf(purgeLimit)); buildCompactor(configProps); assertEquals(purgeLimit * 2, azureStorageCompactor.compactPartition(partitionPath)); }
public int compactPartition(String partitionPath) throws CloudStorageException { if (isShuttingDown()) { logger.info("Skipping compaction of {} due to shut down.", partitionPath); return 0; } Map<String, Long> checkpoints = getCompactionProgress(partitionPath); long now = System.currentTimeMillis(); long queryEndTime = now - retentionPeriodMs; long queryStartTime = now - TimeUnit.DAYS.toMillis(lookbackDays); int totalBlobsPurged = 0; for (String fieldName : compactionFields) { try { long optimizedStartTime = Math.max(queryStartTime, checkpoints.get(fieldName)); int numPurged = compactPartition(partitionPath, fieldName, optimizedStartTime, queryEndTime); logger.info("Purged {} blobs in partition {} based on {}", numPurged, partitionPath, fieldName); totalBlobsPurged += numPurged; } catch (CloudStorageException ex) { logger.error("Compaction failed for partition {} based on {}", partitionPath, fieldName, ex); vcrMetrics.compactionFailureCount.inc(); } } if (isShuttingDown()) { logger.info("Compaction terminated due to shut down."); } return totalBlobsPurged; }
AzureStorageCompactor { public int compactPartition(String partitionPath) throws CloudStorageException { if (isShuttingDown()) { logger.info("Skipping compaction of {} due to shut down.", partitionPath); return 0; } Map<String, Long> checkpoints = getCompactionProgress(partitionPath); long now = System.currentTimeMillis(); long queryEndTime = now - retentionPeriodMs; long queryStartTime = now - TimeUnit.DAYS.toMillis(lookbackDays); int totalBlobsPurged = 0; for (String fieldName : compactionFields) { try { long optimizedStartTime = Math.max(queryStartTime, checkpoints.get(fieldName)); int numPurged = compactPartition(partitionPath, fieldName, optimizedStartTime, queryEndTime); logger.info("Purged {} blobs in partition {} based on {}", numPurged, partitionPath, fieldName); totalBlobsPurged += numPurged; } catch (CloudStorageException ex) { logger.error("Compaction failed for partition {} based on {}", partitionPath, fieldName, ex); vcrMetrics.compactionFailureCount.inc(); } } if (isShuttingDown()) { logger.info("Compaction terminated due to shut down."); } return totalBlobsPurged; } }
AzureStorageCompactor { public int compactPartition(String partitionPath) throws CloudStorageException { if (isShuttingDown()) { logger.info("Skipping compaction of {} due to shut down.", partitionPath); return 0; } Map<String, Long> checkpoints = getCompactionProgress(partitionPath); long now = System.currentTimeMillis(); long queryEndTime = now - retentionPeriodMs; long queryStartTime = now - TimeUnit.DAYS.toMillis(lookbackDays); int totalBlobsPurged = 0; for (String fieldName : compactionFields) { try { long optimizedStartTime = Math.max(queryStartTime, checkpoints.get(fieldName)); int numPurged = compactPartition(partitionPath, fieldName, optimizedStartTime, queryEndTime); logger.info("Purged {} blobs in partition {} based on {}", numPurged, partitionPath, fieldName); totalBlobsPurged += numPurged; } catch (CloudStorageException ex) { logger.error("Compaction failed for partition {} based on {}", partitionPath, fieldName, ex); vcrMetrics.compactionFailureCount.inc(); } } if (isShuttingDown()) { logger.info("Compaction terminated due to shut down."); } return totalBlobsPurged; } AzureStorageCompactor(AzureBlobDataAccessor azureBlobDataAccessor, CosmosDataAccessor cosmosDataAccessor,
      CloudConfig cloudConfig, VcrMetrics vcrMetrics, AzureMetrics azureMetrics); }
AzureStorageCompactor { public int compactPartition(String partitionPath) throws CloudStorageException { if (isShuttingDown()) { logger.info("Skipping compaction of {} due to shut down.", partitionPath); return 0; } Map<String, Long> checkpoints = getCompactionProgress(partitionPath); long now = System.currentTimeMillis(); long queryEndTime = now - retentionPeriodMs; long queryStartTime = now - TimeUnit.DAYS.toMillis(lookbackDays); int totalBlobsPurged = 0; for (String fieldName : compactionFields) { try { long optimizedStartTime = Math.max(queryStartTime, checkpoints.get(fieldName)); int numPurged = compactPartition(partitionPath, fieldName, optimizedStartTime, queryEndTime); logger.info("Purged {} blobs in partition {} based on {}", numPurged, partitionPath, fieldName); totalBlobsPurged += numPurged; } catch (CloudStorageException ex) { logger.error("Compaction failed for partition {} based on {}", partitionPath, fieldName, ex); vcrMetrics.compactionFailureCount.inc(); } } if (isShuttingDown()) { logger.info("Compaction terminated due to shut down."); } return totalBlobsPurged; } AzureStorageCompactor(AzureBlobDataAccessor azureBlobDataAccessor, CosmosDataAccessor cosmosDataAccessor,
      CloudConfig cloudConfig, VcrMetrics vcrMetrics, AzureMetrics azureMetrics); void shutdown(); int compactPartition(String partitionPath); }
AzureStorageCompactor { public int compactPartition(String partitionPath) throws CloudStorageException { if (isShuttingDown()) { logger.info("Skipping compaction of {} due to shut down.", partitionPath); return 0; } Map<String, Long> checkpoints = getCompactionProgress(partitionPath); long now = System.currentTimeMillis(); long queryEndTime = now - retentionPeriodMs; long queryStartTime = now - TimeUnit.DAYS.toMillis(lookbackDays); int totalBlobsPurged = 0; for (String fieldName : compactionFields) { try { long optimizedStartTime = Math.max(queryStartTime, checkpoints.get(fieldName)); int numPurged = compactPartition(partitionPath, fieldName, optimizedStartTime, queryEndTime); logger.info("Purged {} blobs in partition {} based on {}", numPurged, partitionPath, fieldName); totalBlobsPurged += numPurged; } catch (CloudStorageException ex) { logger.error("Compaction failed for partition {} based on {}", partitionPath, fieldName, ex); vcrMetrics.compactionFailureCount.inc(); } } if (isShuttingDown()) { logger.info("Compaction terminated due to shut down."); } return totalBlobsPurged; } AzureStorageCompactor(AzureBlobDataAccessor azureBlobDataAccessor, CosmosDataAccessor cosmosDataAccessor,
      CloudConfig cloudConfig, VcrMetrics vcrMetrics, AzureMetrics azureMetrics); void shutdown(); int compactPartition(String partitionPath);  }