@Test public void testPurge() throws Exception { String blobNameOkStatus = "andromeda"; String blobNameNotFoundStatus = "sirius"; String blobNameErrorStatus = "mutant"; BlobBatch mockBatch = mock(BlobBatch.class); when(mockBatchClient.getBlobBatch()).thenReturn(mockBatch); Response<Void> okResponse = mock(Response.class); when(okResponse.getStatusCode()).thenReturn(202); when(mockBatch.deleteBlob(anyString(), endsWith(blobNameOkStatus))).thenReturn(okResponse); BlobStorageException notFoundException = mock(BlobStorageException.class); when(notFoundException.getStatusCode()).thenReturn(404); Response<Void> notFoundResponse = mock(Response.class); when(notFoundResponse.getStatusCode()).thenThrow(notFoundException); when(mockBatch.deleteBlob(anyString(), endsWith(blobNameNotFoundStatus))).thenReturn(notFoundResponse); BlobStorageException badException = mock(BlobStorageException.class); when(badException.getStatusCode()).thenReturn(503); Response<Void> badResponse = mock(Response.class); when(badResponse.getStatusCode()).thenThrow(badException); when(mockBatch.deleteBlob(anyString(), endsWith(blobNameErrorStatus))).thenReturn(badResponse); List<CloudBlobMetadata> purgeList = new ArrayList<>(); purgeList.add(new CloudBlobMetadata().setId(blobNameOkStatus)); purgeList.add(new CloudBlobMetadata().setId(blobNameNotFoundStatus)); List<CloudBlobMetadata> purgeResponseList = dataAccessor.purgeBlobs(purgeList); assertEquals("Wrong response size", 2, purgeResponseList.size()); assertEquals("Wrong blob name", blobNameOkStatus, purgeResponseList.get(0).getId()); assertEquals("Wrong blob name", blobNameNotFoundStatus, purgeResponseList.get(1).getId()); purgeList.add(new CloudBlobMetadata().setId(blobNameErrorStatus)); try { dataAccessor.purgeBlobs(purgeList); fail("Expected purge to fail"); } catch (BlobStorageException bex) { assertEquals("Unexpected status code", 503, bex.getStatusCode()); } }
public List<CloudBlobMetadata> purgeBlobs(List<CloudBlobMetadata> blobMetadataList) throws BlobStorageException { List<CloudBlobMetadata> deletedBlobs = new ArrayList<>(); List<List<CloudBlobMetadata>> partitionedLists = Utils.partitionList(blobMetadataList, purgeBatchSize); for (List<CloudBlobMetadata> batchOfBlobs : partitionedLists) { BlobBatch blobBatch = blobBatchClient.getBlobBatch(); List<Response<Void>> responseList = new ArrayList<>(); for (CloudBlobMetadata blobMetadata : batchOfBlobs) { BlobLayout blobLayout = blobLayoutStrategy.getDataBlobLayout(blobMetadata); responseList.add(blobBatch.deleteBlob(blobLayout.containerName, blobLayout.blobFilePath)); } blobBatchClient.submitBatchWithResponse(blobBatch, false, batchTimeout, Context.NONE); for (int j = 0; j < responseList.size(); j++) { Response<Void> response = responseList.get(j); CloudBlobMetadata blobMetadata = batchOfBlobs.get(j); try { response.getStatusCode(); } catch (BlobStorageException bex) { int statusCode = bex.getStatusCode(); if (statusCode != HttpURLConnection.HTTP_NOT_FOUND && statusCode != HttpURLConnection.HTTP_GONE) { logger.error("Deleting blob {} got status {}", blobMetadata.getId(), statusCode); throw bex; } } deletedBlobs.add(blobMetadata); } } return deletedBlobs; }
AzureBlobDataAccessor { public List<CloudBlobMetadata> purgeBlobs(List<CloudBlobMetadata> blobMetadataList) throws BlobStorageException { List<CloudBlobMetadata> deletedBlobs = new ArrayList<>(); List<List<CloudBlobMetadata>> partitionedLists = Utils.partitionList(blobMetadataList, purgeBatchSize); for (List<CloudBlobMetadata> batchOfBlobs : partitionedLists) { BlobBatch blobBatch = blobBatchClient.getBlobBatch(); List<Response<Void>> responseList = new ArrayList<>(); for (CloudBlobMetadata blobMetadata : batchOfBlobs) { BlobLayout blobLayout = blobLayoutStrategy.getDataBlobLayout(blobMetadata); responseList.add(blobBatch.deleteBlob(blobLayout.containerName, blobLayout.blobFilePath)); } blobBatchClient.submitBatchWithResponse(blobBatch, false, batchTimeout, Context.NONE); for (int j = 0; j < responseList.size(); j++) { Response<Void> response = responseList.get(j); CloudBlobMetadata blobMetadata = batchOfBlobs.get(j); try { response.getStatusCode(); } catch (BlobStorageException bex) { int statusCode = bex.getStatusCode(); if (statusCode != HttpURLConnection.HTTP_NOT_FOUND && statusCode != HttpURLConnection.HTTP_GONE) { logger.error("Deleting blob {} got status {}", blobMetadata.getId(), statusCode); throw bex; } } deletedBlobs.add(blobMetadata); } } return deletedBlobs; } }
AzureBlobDataAccessor { public List<CloudBlobMetadata> purgeBlobs(List<CloudBlobMetadata> blobMetadataList) throws BlobStorageException { List<CloudBlobMetadata> deletedBlobs = new ArrayList<>(); List<List<CloudBlobMetadata>> partitionedLists = Utils.partitionList(blobMetadataList, purgeBatchSize); for (List<CloudBlobMetadata> batchOfBlobs : partitionedLists) { BlobBatch blobBatch = blobBatchClient.getBlobBatch(); List<Response<Void>> responseList = new ArrayList<>(); for (CloudBlobMetadata blobMetadata : batchOfBlobs) { BlobLayout blobLayout = blobLayoutStrategy.getDataBlobLayout(blobMetadata); responseList.add(blobBatch.deleteBlob(blobLayout.containerName, blobLayout.blobFilePath)); } blobBatchClient.submitBatchWithResponse(blobBatch, false, batchTimeout, Context.NONE); for (int j = 0; j < responseList.size(); j++) { Response<Void> response = responseList.get(j); CloudBlobMetadata blobMetadata = batchOfBlobs.get(j); try { response.getStatusCode(); } catch (BlobStorageException bex) { int statusCode = bex.getStatusCode(); if (statusCode != HttpURLConnection.HTTP_NOT_FOUND && statusCode != HttpURLConnection.HTTP_GONE) { logger.error("Deleting blob {} got status {}", blobMetadata.getId(), statusCode); throw bex; } } deletedBlobs.add(blobMetadata); } } return deletedBlobs; } AzureBlobDataAccessor(CloudConfig cloudConfig, AzureCloudConfig azureCloudConfig,
      AzureBlobLayoutStrategy blobLayoutStrategy, AzureMetrics azureMetrics);   AzureBlobDataAccessor(BlobServiceClient storageClient, BlobBatchClient blobBatchClient, String clusterName,
      AzureMetrics azureMetrics); }
AzureBlobDataAccessor { public List<CloudBlobMetadata> purgeBlobs(List<CloudBlobMetadata> blobMetadataList) throws BlobStorageException { List<CloudBlobMetadata> deletedBlobs = new ArrayList<>(); List<List<CloudBlobMetadata>> partitionedLists = Utils.partitionList(blobMetadataList, purgeBatchSize); for (List<CloudBlobMetadata> batchOfBlobs : partitionedLists) { BlobBatch blobBatch = blobBatchClient.getBlobBatch(); List<Response<Void>> responseList = new ArrayList<>(); for (CloudBlobMetadata blobMetadata : batchOfBlobs) { BlobLayout blobLayout = blobLayoutStrategy.getDataBlobLayout(blobMetadata); responseList.add(blobBatch.deleteBlob(blobLayout.containerName, blobLayout.blobFilePath)); } blobBatchClient.submitBatchWithResponse(blobBatch, false, batchTimeout, Context.NONE); for (int j = 0; j < responseList.size(); j++) { Response<Void> response = responseList.get(j); CloudBlobMetadata blobMetadata = batchOfBlobs.get(j); try { response.getStatusCode(); } catch (BlobStorageException bex) { int statusCode = bex.getStatusCode(); if (statusCode != HttpURLConnection.HTTP_NOT_FOUND && statusCode != HttpURLConnection.HTTP_GONE) { logger.error("Deleting blob {} got status {}", blobMetadata.getId(), statusCode); throw bex; } } deletedBlobs.add(blobMetadata); } } return deletedBlobs; } AzureBlobDataAccessor(CloudConfig cloudConfig, AzureCloudConfig azureCloudConfig,
      AzureBlobLayoutStrategy blobLayoutStrategy, AzureMetrics azureMetrics);   AzureBlobDataAccessor(BlobServiceClient storageClient, BlobBatchClient blobBatchClient, String clusterName,
      AzureMetrics azureMetrics); BlobServiceClient getStorageClient(); boolean uploadIfNotExists(BlobId blobId, long inputLength, CloudBlobMetadata cloudBlobMetadata,
      InputStream blobInputStream); void uploadFile(String containerName, String fileName, InputStream inputStream); boolean downloadFile(String containerName, String fileName, OutputStream outputStream, boolean errorOnNotFound); void downloadBlob(BlobId blobId, OutputStream outputStream); CloudBlobMetadata getBlobMetadata(BlobId blobId); AzureCloudDestination.UpdateResponse updateBlobMetadata(BlobId blobId, Map<String, Object> updateFields,
      CloudUpdateValidator cloudUpdateValidator); List<CloudBlobMetadata> purgeBlobs(List<CloudBlobMetadata> blobMetadataList); }
AzureBlobDataAccessor { public List<CloudBlobMetadata> purgeBlobs(List<CloudBlobMetadata> blobMetadataList) throws BlobStorageException { List<CloudBlobMetadata> deletedBlobs = new ArrayList<>(); List<List<CloudBlobMetadata>> partitionedLists = Utils.partitionList(blobMetadataList, purgeBatchSize); for (List<CloudBlobMetadata> batchOfBlobs : partitionedLists) { BlobBatch blobBatch = blobBatchClient.getBlobBatch(); List<Response<Void>> responseList = new ArrayList<>(); for (CloudBlobMetadata blobMetadata : batchOfBlobs) { BlobLayout blobLayout = blobLayoutStrategy.getDataBlobLayout(blobMetadata); responseList.add(blobBatch.deleteBlob(blobLayout.containerName, blobLayout.blobFilePath)); } blobBatchClient.submitBatchWithResponse(blobBatch, false, batchTimeout, Context.NONE); for (int j = 0; j < responseList.size(); j++) { Response<Void> response = responseList.get(j); CloudBlobMetadata blobMetadata = batchOfBlobs.get(j); try { response.getStatusCode(); } catch (BlobStorageException bex) { int statusCode = bex.getStatusCode(); if (statusCode != HttpURLConnection.HTTP_NOT_FOUND && statusCode != HttpURLConnection.HTTP_GONE) { logger.error("Deleting blob {} got status {}", blobMetadata.getId(), statusCode); throw bex; } } deletedBlobs.add(blobMetadata); } } return deletedBlobs; } AzureBlobDataAccessor(CloudConfig cloudConfig, AzureCloudConfig azureCloudConfig,
      AzureBlobLayoutStrategy blobLayoutStrategy, AzureMetrics azureMetrics);   AzureBlobDataAccessor(BlobServiceClient storageClient, BlobBatchClient blobBatchClient, String clusterName,
      AzureMetrics azureMetrics); BlobServiceClient getStorageClient(); boolean uploadIfNotExists(BlobId blobId, long inputLength, CloudBlobMetadata cloudBlobMetadata,
      InputStream blobInputStream); void uploadFile(String containerName, String fileName, InputStream inputStream); boolean downloadFile(String containerName, String fileName, OutputStream outputStream, boolean errorOnNotFound); void downloadBlob(BlobId blobId, OutputStream outputStream); CloudBlobMetadata getBlobMetadata(BlobId blobId); AzureCloudDestination.UpdateResponse updateBlobMetadata(BlobId blobId, Map<String, Object> updateFields,
      CloudUpdateValidator cloudUpdateValidator); List<CloudBlobMetadata> purgeBlobs(List<CloudBlobMetadata> blobMetadataList);  }