@Test public void testExpire() throws Exception { mockBlobExistence(true); AzureCloudDestination.UpdateResponse response = dataAccessor.updateBlobMetadata(blobId, Collections.singletonMap("expirationTime", expirationTime), dummyCloudUpdateValidator); assertTrue("Expected was updated", response.wasUpdated); assertNotNull("Expected metadata", response.metadata); }
public AzureCloudDestination.UpdateResponse updateBlobMetadata(BlobId blobId, Map<String, Object> updateFields, CloudUpdateValidator cloudUpdateValidator) throws Exception { Objects.requireNonNull(blobId, "BlobId cannot be null"); updateFields.keySet() .forEach(field -> Objects.requireNonNull(updateFields.get(field), String.format("%s cannot be null", field))); try { BlockBlobClient blobClient = getBlockBlobClient(blobId, false); Timer.Context storageTimer = azureMetrics.blobUpdateTime.time(); try { BlobProperties blobProperties = blobClient.getPropertiesWithResponse(defaultRequestConditions, requestTimeout, Context.NONE).getValue(); String etag = blobProperties.getETag(); Map<String, String> metadata = blobProperties.getMetadata(); if (!cloudUpdateValidator.validateUpdate(CloudBlobMetadata.fromMap(metadata), blobId, updateFields)) { return new AzureCloudDestination.UpdateResponse(false, metadata); } Map<String, String> changedFields = updateFields.entrySet() .stream() .filter(entry -> !String.valueOf(entry.getValue()).equals(metadata.get(entry.getKey()))) .collect(Collectors.toMap(Map.Entry::getKey, entry -> String.valueOf(entry.getValue()))); if (changedFields.size() > 0) { changedFields.forEach(metadata::put); if (updateCallback != null) { try { updateCallback.call(); } catch (Exception ex) { logger.error("Error in update callback", ex); } } BlobRequestConditions blobRequestConditions = new BlobRequestConditions().setIfMatch(etag); blobClient.setMetadataWithResponse(metadata, blobRequestConditions, requestTimeout, Context.NONE); return new AzureCloudDestination.UpdateResponse(true, metadata); } else { return new AzureCloudDestination.UpdateResponse(false, metadata); } } finally { storageTimer.stop(); } } catch (BlobStorageException e) { if (isNotFoundError(e.getErrorCode())) { logger.warn("Blob {} not found, cannot update {}.", blobId, updateFields.keySet()); } if (e.getErrorCode() == BlobErrorCode.CONDITION_NOT_MET) { azureMetrics.blobUpdateConflictCount.inc(); } throw e; } }
AzureBlobDataAccessor { public AzureCloudDestination.UpdateResponse updateBlobMetadata(BlobId blobId, Map<String, Object> updateFields, CloudUpdateValidator cloudUpdateValidator) throws Exception { Objects.requireNonNull(blobId, "BlobId cannot be null"); updateFields.keySet() .forEach(field -> Objects.requireNonNull(updateFields.get(field), String.format("%s cannot be null", field))); try { BlockBlobClient blobClient = getBlockBlobClient(blobId, false); Timer.Context storageTimer = azureMetrics.blobUpdateTime.time(); try { BlobProperties blobProperties = blobClient.getPropertiesWithResponse(defaultRequestConditions, requestTimeout, Context.NONE).getValue(); String etag = blobProperties.getETag(); Map<String, String> metadata = blobProperties.getMetadata(); if (!cloudUpdateValidator.validateUpdate(CloudBlobMetadata.fromMap(metadata), blobId, updateFields)) { return new AzureCloudDestination.UpdateResponse(false, metadata); } Map<String, String> changedFields = updateFields.entrySet() .stream() .filter(entry -> !String.valueOf(entry.getValue()).equals(metadata.get(entry.getKey()))) .collect(Collectors.toMap(Map.Entry::getKey, entry -> String.valueOf(entry.getValue()))); if (changedFields.size() > 0) { changedFields.forEach(metadata::put); if (updateCallback != null) { try { updateCallback.call(); } catch (Exception ex) { logger.error("Error in update callback", ex); } } BlobRequestConditions blobRequestConditions = new BlobRequestConditions().setIfMatch(etag); blobClient.setMetadataWithResponse(metadata, blobRequestConditions, requestTimeout, Context.NONE); return new AzureCloudDestination.UpdateResponse(true, metadata); } else { return new AzureCloudDestination.UpdateResponse(false, metadata); } } finally { storageTimer.stop(); } } catch (BlobStorageException e) { if (isNotFoundError(e.getErrorCode())) { logger.warn("Blob {} not found, cannot update {}.", blobId, updateFields.keySet()); } if (e.getErrorCode() == BlobErrorCode.CONDITION_NOT_MET) { azureMetrics.blobUpdateConflictCount.inc(); } throw e; } } }
AzureBlobDataAccessor { public AzureCloudDestination.UpdateResponse updateBlobMetadata(BlobId blobId, Map<String, Object> updateFields, CloudUpdateValidator cloudUpdateValidator) throws Exception { Objects.requireNonNull(blobId, "BlobId cannot be null"); updateFields.keySet() .forEach(field -> Objects.requireNonNull(updateFields.get(field), String.format("%s cannot be null", field))); try { BlockBlobClient blobClient = getBlockBlobClient(blobId, false); Timer.Context storageTimer = azureMetrics.blobUpdateTime.time(); try { BlobProperties blobProperties = blobClient.getPropertiesWithResponse(defaultRequestConditions, requestTimeout, Context.NONE).getValue(); String etag = blobProperties.getETag(); Map<String, String> metadata = blobProperties.getMetadata(); if (!cloudUpdateValidator.validateUpdate(CloudBlobMetadata.fromMap(metadata), blobId, updateFields)) { return new AzureCloudDestination.UpdateResponse(false, metadata); } Map<String, String> changedFields = updateFields.entrySet() .stream() .filter(entry -> !String.valueOf(entry.getValue()).equals(metadata.get(entry.getKey()))) .collect(Collectors.toMap(Map.Entry::getKey, entry -> String.valueOf(entry.getValue()))); if (changedFields.size() > 0) { changedFields.forEach(metadata::put); if (updateCallback != null) { try { updateCallback.call(); } catch (Exception ex) { logger.error("Error in update callback", ex); } } BlobRequestConditions blobRequestConditions = new BlobRequestConditions().setIfMatch(etag); blobClient.setMetadataWithResponse(metadata, blobRequestConditions, requestTimeout, Context.NONE); return new AzureCloudDestination.UpdateResponse(true, metadata); } else { return new AzureCloudDestination.UpdateResponse(false, metadata); } } finally { storageTimer.stop(); } } catch (BlobStorageException e) { if (isNotFoundError(e.getErrorCode())) { logger.warn("Blob {} not found, cannot update {}.", blobId, updateFields.keySet()); } if (e.getErrorCode() == BlobErrorCode.CONDITION_NOT_MET) { azureMetrics.blobUpdateConflictCount.inc(); } throw e; } } AzureBlobDataAccessor(CloudConfig cloudConfig, AzureCloudConfig azureCloudConfig,
      AzureBlobLayoutStrategy blobLayoutStrategy, AzureMetrics azureMetrics);   AzureBlobDataAccessor(BlobServiceClient storageClient, BlobBatchClient blobBatchClient, String clusterName,
      AzureMetrics azureMetrics); }
AzureBlobDataAccessor { public AzureCloudDestination.UpdateResponse updateBlobMetadata(BlobId blobId, Map<String, Object> updateFields, CloudUpdateValidator cloudUpdateValidator) throws Exception { Objects.requireNonNull(blobId, "BlobId cannot be null"); updateFields.keySet() .forEach(field -> Objects.requireNonNull(updateFields.get(field), String.format("%s cannot be null", field))); try { BlockBlobClient blobClient = getBlockBlobClient(blobId, false); Timer.Context storageTimer = azureMetrics.blobUpdateTime.time(); try { BlobProperties blobProperties = blobClient.getPropertiesWithResponse(defaultRequestConditions, requestTimeout, Context.NONE).getValue(); String etag = blobProperties.getETag(); Map<String, String> metadata = blobProperties.getMetadata(); if (!cloudUpdateValidator.validateUpdate(CloudBlobMetadata.fromMap(metadata), blobId, updateFields)) { return new AzureCloudDestination.UpdateResponse(false, metadata); } Map<String, String> changedFields = updateFields.entrySet() .stream() .filter(entry -> !String.valueOf(entry.getValue()).equals(metadata.get(entry.getKey()))) .collect(Collectors.toMap(Map.Entry::getKey, entry -> String.valueOf(entry.getValue()))); if (changedFields.size() > 0) { changedFields.forEach(metadata::put); if (updateCallback != null) { try { updateCallback.call(); } catch (Exception ex) { logger.error("Error in update callback", ex); } } BlobRequestConditions blobRequestConditions = new BlobRequestConditions().setIfMatch(etag); blobClient.setMetadataWithResponse(metadata, blobRequestConditions, requestTimeout, Context.NONE); return new AzureCloudDestination.UpdateResponse(true, metadata); } else { return new AzureCloudDestination.UpdateResponse(false, metadata); } } finally { storageTimer.stop(); } } catch (BlobStorageException e) { if (isNotFoundError(e.getErrorCode())) { logger.warn("Blob {} not found, cannot update {}.", blobId, updateFields.keySet()); } if (e.getErrorCode() == BlobErrorCode.CONDITION_NOT_MET) { azureMetrics.blobUpdateConflictCount.inc(); } throw e; } } 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 AzureCloudDestination.UpdateResponse updateBlobMetadata(BlobId blobId, Map<String, Object> updateFields, CloudUpdateValidator cloudUpdateValidator) throws Exception { Objects.requireNonNull(blobId, "BlobId cannot be null"); updateFields.keySet() .forEach(field -> Objects.requireNonNull(updateFields.get(field), String.format("%s cannot be null", field))); try { BlockBlobClient blobClient = getBlockBlobClient(blobId, false); Timer.Context storageTimer = azureMetrics.blobUpdateTime.time(); try { BlobProperties blobProperties = blobClient.getPropertiesWithResponse(defaultRequestConditions, requestTimeout, Context.NONE).getValue(); String etag = blobProperties.getETag(); Map<String, String> metadata = blobProperties.getMetadata(); if (!cloudUpdateValidator.validateUpdate(CloudBlobMetadata.fromMap(metadata), blobId, updateFields)) { return new AzureCloudDestination.UpdateResponse(false, metadata); } Map<String, String> changedFields = updateFields.entrySet() .stream() .filter(entry -> !String.valueOf(entry.getValue()).equals(metadata.get(entry.getKey()))) .collect(Collectors.toMap(Map.Entry::getKey, entry -> String.valueOf(entry.getValue()))); if (changedFields.size() > 0) { changedFields.forEach(metadata::put); if (updateCallback != null) { try { updateCallback.call(); } catch (Exception ex) { logger.error("Error in update callback", ex); } } BlobRequestConditions blobRequestConditions = new BlobRequestConditions().setIfMatch(etag); blobClient.setMetadataWithResponse(metadata, blobRequestConditions, requestTimeout, Context.NONE); return new AzureCloudDestination.UpdateResponse(true, metadata); } else { return new AzureCloudDestination.UpdateResponse(false, metadata); } } finally { storageTimer.stop(); } } catch (BlobStorageException e) { if (isNotFoundError(e.getErrorCode())) { logger.warn("Blob {} not found, cannot update {}.", blobId, updateFields.keySet()); } if (e.getErrorCode() == BlobErrorCode.CONDITION_NOT_MET) { azureMetrics.blobUpdateConflictCount.inc(); } throw e; } } 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);  }