@Test public void testUpdateNormal() throws Exception { Observable<ResourceResponse<Document>> mockResponse = getMockedObservableForSingleResource(blobMetadata); when(mockumentClient.readDocument(anyString(), any(RequestOptions.class))).thenReturn(mockResponse); when(mockumentClient.replaceDocument(any(), any())).thenReturn(mockResponse); cosmosAccessor.updateMetadata(blobId, Collections.singletonMap(CloudBlobMetadata.FIELD_DELETION_TIME, Long.toString(System.currentTimeMillis()))); assertEquals(1, azureMetrics.documentReadTime.getCount()); assertEquals(1, azureMetrics.documentUpdateTime.getCount()); }
ResourceResponse<Document> updateMetadata(BlobId blobId, Map<String, String> updateFields) throws DocumentClientException { String docLink = getDocumentLink(blobId.getID()); RequestOptions options = getRequestOptions(blobId.getPartition().toPathString()); ResourceResponse<Document> readResponse = executeCosmosAction(() -> asyncDocumentClient.readDocument(docLink, options).toBlocking().single(), azureMetrics.documentReadTime); Document doc = readResponse.getResource(); Map<String, String> fieldsToUpdate = updateFields.entrySet() .stream() .filter(map -> !String.valueOf(updateFields.get(map.getKey())).equals(doc.get(map.getKey()))) .collect(Collectors.toMap(Map.Entry::getKey, map -> String.valueOf(map.getValue()))); if (fieldsToUpdate.size() == 0) { logger.debug("No change in value for {} in blob {}", updateFields.keySet(), blobId.getID()); return null; } if (updateCallback != null) { try { updateCallback.call(); } catch (Exception ex) { logger.error("Error in update callback", ex); } } fieldsToUpdate.forEach((key, value) -> doc.set(key, value)); AccessCondition accessCondition = new AccessCondition(); accessCondition.setCondition(doc.getETag()); options.setAccessCondition(accessCondition); try { return executeCosmosAction(() -> asyncDocumentClient.replaceDocument(doc, options).toBlocking().single(), azureMetrics.documentUpdateTime); } catch (DocumentClientException e) { if (e.getStatusCode() == HttpConstants.StatusCodes.PRECONDITION_FAILED) { azureMetrics.blobUpdateConflictCount.inc(); } throw e; } }
CosmosDataAccessor { ResourceResponse<Document> updateMetadata(BlobId blobId, Map<String, String> updateFields) throws DocumentClientException { String docLink = getDocumentLink(blobId.getID()); RequestOptions options = getRequestOptions(blobId.getPartition().toPathString()); ResourceResponse<Document> readResponse = executeCosmosAction(() -> asyncDocumentClient.readDocument(docLink, options).toBlocking().single(), azureMetrics.documentReadTime); Document doc = readResponse.getResource(); Map<String, String> fieldsToUpdate = updateFields.entrySet() .stream() .filter(map -> !String.valueOf(updateFields.get(map.getKey())).equals(doc.get(map.getKey()))) .collect(Collectors.toMap(Map.Entry::getKey, map -> String.valueOf(map.getValue()))); if (fieldsToUpdate.size() == 0) { logger.debug("No change in value for {} in blob {}", updateFields.keySet(), blobId.getID()); return null; } if (updateCallback != null) { try { updateCallback.call(); } catch (Exception ex) { logger.error("Error in update callback", ex); } } fieldsToUpdate.forEach((key, value) -> doc.set(key, value)); AccessCondition accessCondition = new AccessCondition(); accessCondition.setCondition(doc.getETag()); options.setAccessCondition(accessCondition); try { return executeCosmosAction(() -> asyncDocumentClient.replaceDocument(doc, options).toBlocking().single(), azureMetrics.documentUpdateTime); } catch (DocumentClientException e) { if (e.getStatusCode() == HttpConstants.StatusCodes.PRECONDITION_FAILED) { azureMetrics.blobUpdateConflictCount.inc(); } throw e; } } }
CosmosDataAccessor { ResourceResponse<Document> updateMetadata(BlobId blobId, Map<String, String> updateFields) throws DocumentClientException { String docLink = getDocumentLink(blobId.getID()); RequestOptions options = getRequestOptions(blobId.getPartition().toPathString()); ResourceResponse<Document> readResponse = executeCosmosAction(() -> asyncDocumentClient.readDocument(docLink, options).toBlocking().single(), azureMetrics.documentReadTime); Document doc = readResponse.getResource(); Map<String, String> fieldsToUpdate = updateFields.entrySet() .stream() .filter(map -> !String.valueOf(updateFields.get(map.getKey())).equals(doc.get(map.getKey()))) .collect(Collectors.toMap(Map.Entry::getKey, map -> String.valueOf(map.getValue()))); if (fieldsToUpdate.size() == 0) { logger.debug("No change in value for {} in blob {}", updateFields.keySet(), blobId.getID()); return null; } if (updateCallback != null) { try { updateCallback.call(); } catch (Exception ex) { logger.error("Error in update callback", ex); } } fieldsToUpdate.forEach((key, value) -> doc.set(key, value)); AccessCondition accessCondition = new AccessCondition(); accessCondition.setCondition(doc.getETag()); options.setAccessCondition(accessCondition); try { return executeCosmosAction(() -> asyncDocumentClient.replaceDocument(doc, options).toBlocking().single(), azureMetrics.documentUpdateTime); } catch (DocumentClientException e) { if (e.getStatusCode() == HttpConstants.StatusCodes.PRECONDITION_FAILED) { azureMetrics.blobUpdateConflictCount.inc(); } throw e; } } CosmosDataAccessor(CloudConfig cloudConfig, AzureCloudConfig azureCloudConfig, VcrMetrics vcrMetrics,
      AzureMetrics azureMetrics);   CosmosDataAccessor(AsyncDocumentClient asyncDocumentClient, String cosmosCollectionLink, VcrMetrics vcrMetrics,
      AzureMetrics azureMetrics); }
CosmosDataAccessor { ResourceResponse<Document> updateMetadata(BlobId blobId, Map<String, String> updateFields) throws DocumentClientException { String docLink = getDocumentLink(blobId.getID()); RequestOptions options = getRequestOptions(blobId.getPartition().toPathString()); ResourceResponse<Document> readResponse = executeCosmosAction(() -> asyncDocumentClient.readDocument(docLink, options).toBlocking().single(), azureMetrics.documentReadTime); Document doc = readResponse.getResource(); Map<String, String> fieldsToUpdate = updateFields.entrySet() .stream() .filter(map -> !String.valueOf(updateFields.get(map.getKey())).equals(doc.get(map.getKey()))) .collect(Collectors.toMap(Map.Entry::getKey, map -> String.valueOf(map.getValue()))); if (fieldsToUpdate.size() == 0) { logger.debug("No change in value for {} in blob {}", updateFields.keySet(), blobId.getID()); return null; } if (updateCallback != null) { try { updateCallback.call(); } catch (Exception ex) { logger.error("Error in update callback", ex); } } fieldsToUpdate.forEach((key, value) -> doc.set(key, value)); AccessCondition accessCondition = new AccessCondition(); accessCondition.setCondition(doc.getETag()); options.setAccessCondition(accessCondition); try { return executeCosmosAction(() -> asyncDocumentClient.replaceDocument(doc, options).toBlocking().single(), azureMetrics.documentUpdateTime); } catch (DocumentClientException e) { if (e.getStatusCode() == HttpConstants.StatusCodes.PRECONDITION_FAILED) { azureMetrics.blobUpdateConflictCount.inc(); } throw e; } } CosmosDataAccessor(CloudConfig cloudConfig, AzureCloudConfig azureCloudConfig, VcrMetrics vcrMetrics,
      AzureMetrics azureMetrics);   CosmosDataAccessor(AsyncDocumentClient asyncDocumentClient, String cosmosCollectionLink, VcrMetrics vcrMetrics,
      AzureMetrics azureMetrics); String queryChangeFeed(String requestContinuationToken, int maxFeedSize, List<CloudBlobMetadata> changeFeed,
      String partitionPath, Timer timer); }
CosmosDataAccessor { ResourceResponse<Document> updateMetadata(BlobId blobId, Map<String, String> updateFields) throws DocumentClientException { String docLink = getDocumentLink(blobId.getID()); RequestOptions options = getRequestOptions(blobId.getPartition().toPathString()); ResourceResponse<Document> readResponse = executeCosmosAction(() -> asyncDocumentClient.readDocument(docLink, options).toBlocking().single(), azureMetrics.documentReadTime); Document doc = readResponse.getResource(); Map<String, String> fieldsToUpdate = updateFields.entrySet() .stream() .filter(map -> !String.valueOf(updateFields.get(map.getKey())).equals(doc.get(map.getKey()))) .collect(Collectors.toMap(Map.Entry::getKey, map -> String.valueOf(map.getValue()))); if (fieldsToUpdate.size() == 0) { logger.debug("No change in value for {} in blob {}", updateFields.keySet(), blobId.getID()); return null; } if (updateCallback != null) { try { updateCallback.call(); } catch (Exception ex) { logger.error("Error in update callback", ex); } } fieldsToUpdate.forEach((key, value) -> doc.set(key, value)); AccessCondition accessCondition = new AccessCondition(); accessCondition.setCondition(doc.getETag()); options.setAccessCondition(accessCondition); try { return executeCosmosAction(() -> asyncDocumentClient.replaceDocument(doc, options).toBlocking().single(), azureMetrics.documentUpdateTime); } catch (DocumentClientException e) { if (e.getStatusCode() == HttpConstants.StatusCodes.PRECONDITION_FAILED) { azureMetrics.blobUpdateConflictCount.inc(); } throw e; } } CosmosDataAccessor(CloudConfig cloudConfig, AzureCloudConfig azureCloudConfig, VcrMetrics vcrMetrics,
      AzureMetrics azureMetrics);   CosmosDataAccessor(AsyncDocumentClient asyncDocumentClient, String cosmosCollectionLink, VcrMetrics vcrMetrics,
      AzureMetrics azureMetrics); String queryChangeFeed(String requestContinuationToken, int maxFeedSize, List<CloudBlobMetadata> changeFeed,
      String partitionPath, Timer timer); static final String COSMOS_LAST_UPDATED_COLUMN; }