NIFI-12271 Fix PutAzureBlobStorage_v12 rollback on failure with FileR…

…esourceService This closes apache#7930 Signed-off-by: David Handermann <[email protected]>
exceptionfactory · Oct 26, 2023 · a3e4f89 · a3e4f89
1 parent c4ff8de
commit a3e4f89
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 1 deletion.
diff --git a/...ssors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureBlobStorage_v12.java b/...ssors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureBlobStorage_v12.java
@@ -166,10 +166,10 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
         final String blobName = context.getProperty(BLOB_NAME).evaluateAttributeExpressions(flowFile).getValue();
         final AzureStorageConflictResolutionStrategy conflictResolution = AzureStorageConflictResolutionStrategy.valueOf(context.getProperty(CONFLICT_RESOLUTION).getValue());
         final ResourceTransferSource resourceTransferSource = ResourceTransferSource.valueOf(context.getProperty(RESOURCE_TRANSFER_SOURCE).getValue());
-        final Optional<FileResource> fileResourceFound = getFileResource(resourceTransferSource, context, flowFile.getAttributes());
 
         long startNanos = System.nanoTime();
         try {
+            final Optional<FileResource> fileResourceFound = getFileResource(resourceTransferSource, context, flowFile.getAttributes());
             BlobServiceClient storageClient = getStorageClient(context, flowFile);
             BlobContainerClient containerClient = storageClient.getBlobContainerClient(containerName);
             if (createContainer && !containerClient.exists()) {

diff --git a/...ors/src/test/java/org/apache/nifi/processors/azure/storage/ITPutAzureBlobStorage_v12.java b/...ors/src/test/java/org/apache/nifi/processors/azure/storage/ITPutAzureBlobStorage_v12.java
@@ -284,6 +284,31 @@ public void testPutBlobFromLocalFile() throws Exception {
         assertProvenanceEvents();
     }
 
+    @Test
+    public void testPutBlobFromNonExistentLocalFile() throws Exception {
+        String attributeName = "file.path";
+
+        String serviceId = FileResourceService.class.getSimpleName();
+        FileResourceService service = new StandardFileResourceService();
+        runner.addControllerService(serviceId, service);
+        runner.setProperty(service, StandardFileResourceService.FILE_PATH, String.format("${%s}", attributeName));
+        runner.enableControllerService(service);
+
+        runner.setProperty(ResourceTransferProperties.RESOURCE_TRANSFER_SOURCE, ResourceTransferSource.FILE_RESOURCE_SERVICE.getValue());
+        runner.setProperty(ResourceTransferProperties.FILE_RESOURCE_SERVICE, serviceId);
+
+        String filePath = "nonexistent.txt";
+
+        Map<String, String> attributes = new HashMap<>();
+        attributes.put(attributeName, filePath);
+
+        runProcessor(EMPTY_CONTENT, attributes);
+
+        runner.assertAllFlowFilesTransferred(PutAzureBlobStorage_v12.REL_FAILURE, 1);
+
+        assertProvenanceEvents();
+    }
+
 
     private void runProcessor(byte[] data) {
         runProcessor(data, Collections.emptyMap());