### Question:
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);  }### Answer:
@Test public void testCompaction() throws Exception { int expectedNumQUeries = numQueryBuckets * 2; int expectedPurged = numBlobsPerQuery * expectedNumQUeries; assertEquals(expectedPurged, azureStorageCompactor.compactPartition(partitionPath)); verify(mockumentClient, times(expectedNumQUeries)).queryDocuments(eq(collectionLink), any(SqlQuerySpec.class), any()); verify(mockumentClient, times(expectedNumQUeries)).executeStoredProcedure( eq(collectionLink + CosmosDataAccessor.BULK_DELETE_SPROC), any(), any()); verify(mockBlobBatchClient, times(expectedNumQUeries)).submitBatchWithResponse(any(BlobBatch.class), anyBoolean(), any(), any()); }
@Test public void testCompactionProceedsOnCheckpointNotFound() throws Exception { mockCheckpointDownload(false, null); int expectedNumQUeries = numQueryBuckets * 2; int expectedPurged = numBlobsPerQuery * expectedNumQUeries; assertEquals(expectedPurged, azureStorageCompactor.compactPartition(partitionPath)); verify(mockumentClient, times(expectedNumQUeries)).queryDocuments(eq(collectionLink), any(SqlQuerySpec.class), any()); }
@Test public void testCompactionFailsOnCheckpointReadError() throws Exception { BlobStorageException ex = mockStorageException(BlobErrorCode.INTERNAL_ERROR); when(mockBlockBlobClient.downloadWithResponse(any(), any(), any(), any(), anyBoolean(), any(), any())).thenThrow( ex); try { azureStorageCompactor.compactPartition(partitionPath); fail("Expected compaction to fail"); } catch (CloudStorageException cse) { } }
@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)); }