This is an automated email from the ASF dual-hosted git repository.
pvillard31 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new fb432a2db5a NIFI-7307 Adding support for setting and retrieving Azure
Blob Storage metadata and tags (#10939)
fb432a2db5a is described below
commit fb432a2db5a766c5fb2929625569ceb998c6f5e9
Author: pkelly-nifi <[email protected]>
AuthorDate: Thu Jul 16 08:45:00 2026 -0400
NIFI-7307 Adding support for setting and retrieving Azure Blob Storage
metadata and tags (#10939)
---
...ctFetchAzureBlobStoragePropertiesProcessor.java | 122 ++++++++
.../storage/FetchAzureBlobStorageMetadata.java | 50 +++
.../azure/storage/FetchAzureBlobStorageTags.java | 50 +++
.../azure/storage/PutAzureBlobStorage_v12.java | 100 +++++-
.../azure/storage/utils/BlobAttributes.java | 4 +
.../services/org.apache.nifi.processor.Processor | 2 +
.../storage/TestFetchAzureBlobStorageMetadata.java | 271 ++++++++++++++++
.../storage/TestFetchAzureBlobStorageTags.java | 237 ++++++++++++++
.../azure/storage/TestPutAzureBlobStorage_v12.java | 345 +++++++++++++++++++++
9 files changed, 1179 insertions(+), 2 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/AbstractFetchAzureBlobStoragePropertiesProcessor.java
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/AbstractFetchAzureBlobStoragePropertiesProcessor.java
new file mode 100644
index 00000000000..b46f9ca1ea6
--- /dev/null
+++
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/AbstractFetchAzureBlobStoragePropertiesProcessor.java
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.azure.storage;
+
+import com.azure.storage.blob.BlobClient;
+import com.azure.storage.blob.BlobContainerClient;
+import com.azure.storage.blob.BlobServiceClient;
+import com.azure.storage.blob.models.BlobErrorCode;
+import com.azure.storage.blob.models.BlobStorageException;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processors.azure.AbstractAzureBlobProcessor_v12;
+import org.apache.nifi.processors.azure.storage.utils.AzureStorageUtils;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static
org.apache.nifi.processors.azure.storage.utils.AzureStorageUtils.BLOB_STORAGE_CREDENTIALS_SERVICE;
+import static
org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR_NAME_BLOBNAME;
+import static
org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR_NAME_CONTAINER;
+
+public abstract class AbstractFetchAzureBlobStoragePropertiesProcessor extends
AbstractAzureBlobProcessor_v12 {
+
+ public static final PropertyDescriptor CONTAINER = new
PropertyDescriptor.Builder()
+ .fromPropertyDescriptor(AzureStorageUtils.CONTAINER)
+ .defaultValue(String.format("${%s}", ATTR_NAME_CONTAINER))
+ .build();
+
+ public static final PropertyDescriptor BLOB_NAME = new
PropertyDescriptor.Builder()
+ .fromPropertyDescriptor(AbstractAzureBlobProcessor_v12.BLOB_NAME)
+ .defaultValue(String.format("${%s}", ATTR_NAME_BLOBNAME))
+ .build();
+
+ private static final List<PropertyDescriptor> PROPERTIES = List.of(
+ BLOB_STORAGE_CREDENTIALS_SERVICE,
+ CONTAINER,
+ BLOB_NAME,
+ AzureStorageUtils.PROXY_CONFIGURATION_SERVICE
+ );
+
+ public static final Relationship REL_FOUND = new Relationship.Builder()
+ .name("found")
+ .description("A blob with the supplied name was found in the
container")
+ .build();
+
+ public static final Relationship REL_NOT_FOUND = new Relationship.Builder()
+ .name("not found")
+ .description("No blob was found with the supplied name in the
container")
+ .build();
+
+ private static final Set<Relationship> RELATIONSHIPS = Set.of(REL_FOUND,
REL_NOT_FOUND, REL_FAILURE);
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ protected abstract String getAttributePrefix();
+
+ protected abstract Map<String, String> fetchProperties(BlobClient
blobClient);
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession
session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final String containerName =
context.getProperty(CONTAINER).evaluateAttributeExpressions(flowFile).getValue();
+ final String blobName =
context.getProperty(BLOB_NAME).evaluateAttributeExpressions(flowFile).getValue();
+
+ try {
+ final BlobServiceClient storageClient = getStorageClient(context,
flowFile);
+ final BlobContainerClient containerClient =
storageClient.getBlobContainerClient(containerName);
+ final BlobClient blobClient =
containerClient.getBlobClient(blobName);
+
+ final Map<String, String> newAttributes = new HashMap<>();
+ fetchProperties(blobClient).forEach((key, value) ->
+ newAttributes.put(getAttributePrefix().formatted(key), value)
+ );
+
+ flowFile = session.putAllAttributes(flowFile, newAttributes);
+ final String eventDetails = String.format("Attributes added [%s]",
String.join(", ", newAttributes.keySet()));
+ session.getProvenanceReporter().modifyAttributes(flowFile,
eventDetails);
+ session.transfer(flowFile, REL_FOUND);
+ } catch (final BlobStorageException e) {
+ if (e.getErrorCode() == BlobErrorCode.BLOB_NOT_FOUND) {
+ getLogger().warn("Specified blob ({}) does not exist, routing
to not found.", blobName);
+ session.transfer(flowFile, REL_NOT_FOUND);
+ } else {
+ getLogger().error("Failed to retrieve properties for the
specified blob ({}) from Azure Blob Storage. Routing to failure", blobName, e);
+ flowFile = session.penalize(flowFile);
+ session.transfer(flowFile, REL_FAILURE);
+ }
+ }
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/FetchAzureBlobStorageMetadata.java
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/FetchAzureBlobStorageMetadata.java
new file mode 100644
index 00000000000..9794356651b
--- /dev/null
+++
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/FetchAzureBlobStorageMetadata.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.azure.storage;
+
+import com.azure.storage.blob.BlobClient;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+
+import java.util.Map;
+import java.util.Optional;
+
+@Tags({"azure", "microsoft", "cloud", "storage", "blob"})
+@SeeAlso({ListAzureBlobStorage_v12.class, FetchAzureBlobStorage_v12.class,
PutAzureBlobStorage_v12.class,
+ CopyAzureBlobStorage_v12.class, DeleteAzureBlobStorage_v12.class,
FetchAzureBlobStorageTags.class})
+@CapabilityDescription("Retrieves user metadata from the specified blob from
Azure Blob Storage. The processor uses Azure Blob Storage client library v12.")
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@WritesAttributes({@WritesAttribute(attribute =
"azure.user.metadata.<metadata-key>", description = "The value of the retrieved
metadata")})
+public class FetchAzureBlobStorageMetadata extends
AbstractFetchAzureBlobStoragePropertiesProcessor {
+
+ private static final String ATTRIBUTE_PREFIX = "azure.user.metadata.%s";
+
+ @Override
+ protected String getAttributePrefix() {
+ return ATTRIBUTE_PREFIX;
+ }
+
+ @Override
+ protected Map<String, String> fetchProperties(final BlobClient blobClient)
{
+ return
Optional.ofNullable(blobClient.getProperties().getMetadata()).orElse(Map.of());
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/FetchAzureBlobStorageTags.java
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/FetchAzureBlobStorageTags.java
new file mode 100644
index 00000000000..0f663bbe625
--- /dev/null
+++
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/FetchAzureBlobStorageTags.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.azure.storage;
+
+import com.azure.storage.blob.BlobClient;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+
+import java.util.Map;
+import java.util.Optional;
+
+@Tags({"azure", "microsoft", "cloud", "storage", "blob"})
+@SeeAlso({ListAzureBlobStorage_v12.class, FetchAzureBlobStorage_v12.class,
PutAzureBlobStorage_v12.class,
+ CopyAzureBlobStorage_v12.class, DeleteAzureBlobStorage_v12.class,
FetchAzureBlobStorageMetadata.class})
+@CapabilityDescription("Retrieves tags from the specified blob from Azure Blob
Storage. The processor uses Azure Blob Storage client library v12.")
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@WritesAttributes({@WritesAttribute(attribute = "azure.tag.<tag-key>",
description = "The value of the retrieved tag")})
+public class FetchAzureBlobStorageTags extends
AbstractFetchAzureBlobStoragePropertiesProcessor {
+
+ private static final String ATTRIBUTE_PREFIX = "azure.tag.%s";
+
+ @Override
+ protected String getAttributePrefix() {
+ return ATTRIBUTE_PREFIX;
+ }
+
+ @Override
+ protected Map<String, String> fetchProperties(final BlobClient blobClient)
{
+ return Optional.ofNullable(blobClient.getTags()).orElse(Map.of());
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureBlobStorage_v12.java
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureBlobStorage_v12.java
index 50c0a4d9d9f..07d5f814248 100644
---
a/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureBlobStorage_v12.java
+++
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureBlobStorage_v12.java
@@ -29,6 +29,8 @@ import com.azure.storage.blob.models.BlobType;
import com.azure.storage.blob.models.BlockBlobItem;
import com.azure.storage.blob.models.ParallelTransferOptions;
import com.azure.storage.blob.options.BlobParallelUploadOptions;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.DynamicProperty;
import org.apache.nifi.annotation.behavior.InputRequirement;
import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
import org.apache.nifi.annotation.behavior.WritesAttribute;
@@ -36,15 +38,18 @@ import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.fileresource.service.api.FileResource;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.migration.PropertyConfiguration;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.processors.azure.AbstractAzureBlobProcessor_v12;
import org.apache.nifi.processors.azure.ClientSideEncryptionSupport;
import org.apache.nifi.processors.azure.storage.utils.AzureStorageUtils;
@@ -59,6 +64,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import static com.azure.core.http.ContentType.APPLICATION_OCTET_STREAM;
import static com.azure.core.util.FluxUtil.toFluxByteBuffer;
@@ -76,6 +82,7 @@ import static
org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR
import static
org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR_DESCRIPTION_MIME_TYPE;
import static
org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR_DESCRIPTION_PRIMARY_URI;
import static
org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR_DESCRIPTION_TIMESTAMP;
+import static
org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR_DESCRIPTION_USER_METADATA;
import static
org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR_NAME_BLOBNAME;
import static
org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR_NAME_BLOBTYPE;
import static
org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR_NAME_CONTAINER;
@@ -87,14 +94,19 @@ import static
org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR
import static
org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR_NAME_MIME_TYPE;
import static
org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR_NAME_PRIMARY_URI;
import static
org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR_NAME_TIMESTAMP;
+import static
org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR_NAME_USER_METADATA;
import static
org.apache.nifi.processors.transfer.ResourceTransferProperties.FILE_RESOURCE_SERVICE;
import static
org.apache.nifi.processors.transfer.ResourceTransferProperties.RESOURCE_TRANSFER_SOURCE;
import static
org.apache.nifi.processors.transfer.ResourceTransferUtils.getFileResource;
@Tags({"azure", "microsoft", "cloud", "storage", "blob"})
@SeeAlso({ListAzureBlobStorage_v12.class, FetchAzureBlobStorage_v12.class,
DeleteAzureBlobStorage_v12.class,
- CopyAzureBlobStorage_v12.class})
+ CopyAzureBlobStorage_v12.class, FetchAzureBlobStorageMetadata.class,
FetchAzureBlobStorageTags.class})
@CapabilityDescription("Puts content into a blob on Azure Blob Storage. The
processor uses Azure Blob Storage client library v12.")
+@DynamicProperty(name = "The name of a User-Defined Metadata field to add to
the blob",
+ value = "The value of a User-Defined Metadata field to add to the
blob",
+ description = "Allows user-defined metadata to be added to the blob as
key/value pairs",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
@InputRequirement(Requirement.INPUT_REQUIRED)
@WritesAttributes({@WritesAttribute(attribute = ATTR_NAME_CONTAINER,
description = ATTR_DESCRIPTION_CONTAINER),
@WritesAttribute(attribute = ATTR_NAME_BLOBNAME, description =
ATTR_DESCRIPTION_BLOBNAME),
@@ -106,9 +118,33 @@ import static
org.apache.nifi.processors.transfer.ResourceTransferUtils.getFileR
@WritesAttribute(attribute = ATTR_NAME_TIMESTAMP, description =
ATTR_DESCRIPTION_TIMESTAMP),
@WritesAttribute(attribute = ATTR_NAME_LENGTH, description =
ATTR_DESCRIPTION_LENGTH),
@WritesAttribute(attribute = ATTR_NAME_ERROR_CODE, description =
ATTR_DESCRIPTION_ERROR_CODE),
- @WritesAttribute(attribute = ATTR_NAME_IGNORED, description =
ATTR_DESCRIPTION_IGNORED)})
+ @WritesAttribute(attribute = ATTR_NAME_IGNORED, description =
ATTR_DESCRIPTION_IGNORED),
+ @WritesAttribute(attribute = ATTR_NAME_USER_METADATA, description =
ATTR_DESCRIPTION_USER_METADATA)})
public class PutAzureBlobStorage_v12 extends AbstractAzureBlobProcessor_v12
implements ClientSideEncryptionSupport {
+ public static final PropertyDescriptor BLOB_TAG_PREFIX = new
PropertyDescriptor.Builder()
+ .name("Blob Tag Prefix")
+ .description("""
+ Specifies the prefix which would be scanned against the
incoming FlowFile's attributes and the matching attribute's
+ name and value would be considered as the outgoing Azure
blob's Tag name and Tag value respectively. For Ex: If the
+ incoming FlowFile carries the attributes tagAzurecountry,
tagAzurePII, the tag prefix to be specified would be 'tagAzure'.
+ """)
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
+
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .build();
+
+ public static final PropertyDescriptor REMOVE_TAG_PREFIX = new
PropertyDescriptor.Builder()
+ .name("Remove Tag Prefix")
+ .description("""
+ If set to 'True', the value provided for '%s' will be
removed from
+ the attribute(s) and then considered as the Tag name. For
ex: If the incoming FlowFile carries the attributes tagAzurecountry,
+ tagAzurePII and the prefix is set to 'tagAzure' then the
corresponding tag names would be 'country' and 'PII'.
+ """.formatted(BLOB_TAG_PREFIX.getDisplayName()))
+ .allowableValues(new AllowableValue("true", "True"), new
AllowableValue("false", "False"))
+ .defaultValue("false")
+ .build();
+
private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS =
List.of(
BLOB_STORAGE_CREDENTIALS_SERVICE,
AzureStorageUtils.CONTAINER,
@@ -116,6 +152,8 @@ public class PutAzureBlobStorage_v12 extends
AbstractAzureBlobProcessor_v12 impl
AzureStorageUtils.CONFLICT_RESOLUTION,
BLOB_NAME,
CONTENT_MD5,
+ BLOB_TAG_PREFIX,
+ REMOVE_TAG_PREFIX,
RESOURCE_TRANSFER_SOURCE,
FILE_RESOURCE_SERVICE,
AzureStorageUtils.PROXY_CONFIGURATION_SERVICE,
@@ -124,6 +162,16 @@ public class PutAzureBlobStorage_v12 extends
AbstractAzureBlobProcessor_v12 impl
CSE_LOCAL_KEY
);
+ @Override
+ protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final
String propertyDescriptorName) {
+ return new PropertyDescriptor.Builder()
+ .name(propertyDescriptorName)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .dynamic(true)
+ .build();
+ }
+
@Override
protected Collection<ValidationResult> customValidate(ValidationContext
validationContext) {
final List<ValidationResult> results = new
ArrayList<>(super.customValidate(validationContext));
@@ -187,6 +235,14 @@ public class PutAzureBlobStorage_v12 extends
AbstractAzureBlobProcessor_v12 impl
blobParallelUploadOptions.setParallelTransferOptions(parallelTransferOptions);
blobParallelUploadOptions.setRequestConditions(blobRequestConditions);
+ if (context.getProperty(BLOB_TAG_PREFIX).isSet()) {
+
blobParallelUploadOptions.setTags(getObjectTags(context, flowFile));
+ }
+ final Map<String, String> userMetadata =
getUserMetadata(context, flowFile);
+ if (!userMetadata.isEmpty()) {
+ blobParallelUploadOptions.setMetadata(userMetadata);
+ }
+
final String contentMd5 =
context.getProperty(CONTENT_MD5).evaluateAttributeExpressions(sourceFlowFile).getValue();
if (contentMd5 != null) {
final byte[] md5Bytes = convertMd5ToBytes(contentMd5);
@@ -201,6 +257,12 @@ public class PutAzureBlobStorage_v12 extends
AbstractAzureBlobProcessor_v12 impl
if (ignore) {
attributes.put(ATTR_NAME_IGNORED, "false");
}
+ final String userMetadataAttributeValue =
userMetadata.entrySet().stream()
+ .map(entry -> String.format("%s=%s",
entry.getKey(), entry.getValue()))
+ .collect(Collectors.joining("\n"));
+ if (StringUtils.isNotBlank(userMetadataAttributeValue)) {
+ attributes.put(ATTR_NAME_USER_METADATA,
userMetadataAttributeValue);
+ }
}
} catch (BlobStorageException e) {
final BlobErrorCode errorCode = e.getErrorCode();
@@ -245,4 +307,38 @@ public class PutAzureBlobStorage_v12 extends
AbstractAzureBlobProcessor_v12 impl
attributes.put(ATTR_NAME_LANG, null);
attributes.put(ATTR_NAME_MIME_TYPE, APPLICATION_OCTET_STREAM);
}
+
+ private Map<String, String> getObjectTags(ProcessContext context, FlowFile
flowFile) {
+ final String prefix =
context.getProperty(BLOB_TAG_PREFIX).evaluateAttributeExpressions(flowFile).getValue();
+ final Map<String, String> objectTags = new HashMap<>();
+ final Map<String, String> attributesMap = flowFile.getAttributes();
+
+ attributesMap.entrySet().stream()
+ .filter(attribute -> attribute.getKey().startsWith(prefix))
+ .forEach(attribute -> {
+ String tagKey = attribute.getKey();
+ String tagValue = attribute.getValue();
+
+ if (context.getProperty(REMOVE_TAG_PREFIX).asBoolean()) {
+ tagKey = tagKey.substring(prefix.length());
+ }
+ objectTags.put(tagKey, tagValue);
+ });
+
+ return objectTags;
+ }
+
+ private Map<String, String> getUserMetadata(ProcessContext context,
FlowFile flowFile) {
+ final Map<String, String> userMetadata = new HashMap<>();
+
+ for (final Map.Entry<PropertyDescriptor, String> entry :
context.getProperties().entrySet()) {
+ if (entry.getKey().isDynamic()) {
+ final String value = context.getProperty(
+
entry.getKey()).evaluateAttributeExpressions(flowFile).getValue();
+ userMetadata.put(entry.getKey().getName(), value);
+ }
+ }
+
+ return userMetadata;
+ }
}
diff --git
a/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/BlobAttributes.java
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/BlobAttributes.java
index 7ce30c1da7c..7d64406f587 100644
---
a/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/BlobAttributes.java
+++
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/BlobAttributes.java
@@ -51,4 +51,8 @@ public final class BlobAttributes {
public static final String ATTR_NAME_IGNORED = "azure.ignored";
public static final String ATTR_DESCRIPTION_IGNORED = "When Conflict
Resolution Strategy is 'ignore', " +
"this property will be true/false depending on whether the blob
was ignored.";
+
+ public static final String ATTR_NAME_USER_METADATA = "azure.usermetadata";
+ public static final String ATTR_DESCRIPTION_USER_METADATA = "A
human-readable form of the User Metadata of " +
+ "the blob, if any was set";
}
diff --git
a/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
index f7ef2f37070..de8b21e3b0d 100644
---
a/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
+++
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
@@ -27,6 +27,8 @@
org.apache.nifi.processors.azure.storage.FetchAzureBlobStorage_v12
org.apache.nifi.processors.azure.storage.PutAzureBlobStorage_v12
org.apache.nifi.processors.azure.storage.DeleteAzureBlobStorage_v12
org.apache.nifi.processors.azure.storage.CopyAzureBlobStorage_v12
+org.apache.nifi.processors.azure.storage.FetchAzureBlobStorageMetadata
+org.apache.nifi.processors.azure.storage.FetchAzureBlobStorageTags
org.apache.nifi.processors.azure.storage.MoveAzureDataLakeStorage
org.apache.nifi.processors.azure.storage.queue.GetAzureQueueStorage_v12
org.apache.nifi.processors.azure.storage.queue.PutAzureQueueStorage_v12
diff --git
a/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/TestFetchAzureBlobStorageMetadata.java
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/TestFetchAzureBlobStorageMetadata.java
new file mode 100644
index 00000000000..9453e98b742
--- /dev/null
+++
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/TestFetchAzureBlobStorageMetadata.java
@@ -0,0 +1,271 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.azure.storage;
+
+import com.azure.storage.blob.BlobClient;
+import com.azure.storage.blob.BlobContainerClient;
+import com.azure.storage.blob.BlobServiceClient;
+import com.azure.storage.blob.models.BlobErrorCode;
+import com.azure.storage.blob.models.BlobProperties;
+import com.azure.storage.blob.models.BlobStorageException;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.provenance.ProvenanceEventRecord;
+import org.apache.nifi.provenance.ProvenanceEventType;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static
org.apache.nifi.processors.azure.storage.utils.AzureStorageUtils.BLOB_STORAGE_CREDENTIALS_SERVICE;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TestFetchAzureBlobStorageMetadata {
+
+ private static final String CONTAINER_NAME = "test-container";
+ private static final String BLOB_NAME = "test-blob";
+
+ private TestRunner runner;
+ private BlobServiceClient storageClient;
+ private BlobClient blobClient;
+ private BlobProperties blobProperties;
+
+ @BeforeEach
+ void setUp() {
+ storageClient = mock(BlobServiceClient.class);
+ final BlobContainerClient containerClient =
mock(BlobContainerClient.class);
+ blobClient = mock(BlobClient.class);
+ blobProperties = mock(BlobProperties.class);
+
+
when(storageClient.getBlobContainerClient(CONTAINER_NAME)).thenReturn(containerClient);
+ when(containerClient.getBlobClient(BLOB_NAME)).thenReturn(blobClient);
+ when(blobClient.getProperties()).thenReturn(blobProperties);
+
+ final FetchAzureBlobStorageMetadata processor = new
FetchAzureBlobStorageMetadata() {
+ @Override
+ protected BlobServiceClient getStorageClient(PropertyContext
context, FlowFile flowFile) {
+ return storageClient;
+ }
+
+ @Override
+ protected List<PropertyDescriptor>
getSupportedPropertyDescriptors() {
+ return super.getSupportedPropertyDescriptors().stream()
+ .filter(pd ->
!pd.equals(BLOB_STORAGE_CREDENTIALS_SERVICE))
+ .toList();
+ }
+ };
+
+ runner = TestRunners.newTestRunner(processor);
+
runner.setProperty(AbstractFetchAzureBlobStoragePropertiesProcessor.CONTAINER,
CONTAINER_NAME);
+
runner.setProperty(AbstractFetchAzureBlobStoragePropertiesProcessor.BLOB_NAME,
BLOB_NAME);
+ }
+
+ @Test
+ void testSuccessfulMetadataRetrieval() {
+ final Map<String, String> metadata = Map.of(
+ "author", "jane-doe",
+ "source-system", "erp",
+ "processing-date", "2024-01-15"
+ );
+ when(blobProperties.getMetadata()).thenReturn(metadata);
+
+ runner.enqueue("");
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(
+ AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND, 1);
+
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(
+
AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND).getFirst();
+
+ assertEquals("jane-doe",
flowFile.getAttribute("azure.user.metadata.author"));
+ assertEquals("erp",
flowFile.getAttribute("azure.user.metadata.source-system"));
+ assertEquals("2024-01-15",
flowFile.getAttribute("azure.user.metadata.processing-date"));
+ }
+
+ @Test
+ void testEmptyMetadataRetrieval() {
+ when(blobProperties.getMetadata()).thenReturn(Map.of());
+
+ runner.enqueue("");
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(
+ AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND, 1);
+
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(
+
AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND).getFirst();
+
+ flowFile.getAttributes().forEach((key, value) ->
+ assertFalse(key.startsWith("azure.user.metadata."),
+ "No metadata attributes should exist when blob has none,
found: " + key)
+ );
+ }
+
+ @Test
+ void testBlobNotFound() {
+ BlobStorageException exception =
mockBlobStorageException(BlobErrorCode.BLOB_NOT_FOUND);
+ when(blobClient.getProperties()).thenThrow(exception);
+
+ runner.enqueue("");
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(
+
AbstractFetchAzureBlobStoragePropertiesProcessor.REL_NOT_FOUND, 1);
+ }
+
+ @Test
+ void testOtherBlobStorageExceptionRoutesToFailure() {
+ BlobStorageException exception =
mockBlobStorageException(BlobErrorCode.AUTHORIZATION_FAILURE);
+ when(blobClient.getProperties()).thenThrow(exception);
+
+ runner.enqueue("");
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(
+ AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FAILURE,
1);
+
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(
+
AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FAILURE).getFirst();
+ assertTrue(flowFile.isPenalized(), "FlowFile should be penalized on
failure");
+ }
+
+ @Test
+ void testContainerAndBlobNameFromFlowFileAttributes() {
+
runner.setProperty(AbstractFetchAzureBlobStoragePropertiesProcessor.CONTAINER,
+ "${azure.container}");
+
runner.setProperty(AbstractFetchAzureBlobStoragePropertiesProcessor.BLOB_NAME,
+ "${azure.blobname}");
+
+ final String dynamicContainer = "other-container";
+ final String dynamicBlob = "other-blob";
+
+ final BlobContainerClient otherContainerClient =
mock(BlobContainerClient.class);
+ final BlobClient otherBlobClient = mock(BlobClient.class);
+ final BlobProperties otherBlobProperties = mock(BlobProperties.class);
+
+ when(storageClient.getBlobContainerClient(dynamicContainer))
+ .thenReturn(otherContainerClient);
+ when(otherContainerClient.getBlobClient(dynamicBlob))
+ .thenReturn(otherBlobClient);
+ when(otherBlobClient.getProperties()).thenReturn(otherBlobProperties);
+ when(otherBlobProperties.getMetadata()).thenReturn(Map.of("origin",
"external"));
+
+ runner.enqueue("", Map.of(
+ "azure.container", dynamicContainer,
+ "azure.blobname", dynamicBlob
+ ));
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(
+ AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND, 1);
+
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(
+
AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND).getFirst();
+ assertEquals("external",
flowFile.getAttribute("azure.user.metadata.origin"));
+ }
+
+ @Test
+ void testProvenanceEventOnFound() {
+ when(blobProperties.getMetadata()).thenReturn(Map.of("key", "value"));
+
+ runner.enqueue("");
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(
+ AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND, 1);
+
+ final ProvenanceEventRecord modifyEvent =
runner.getProvenanceEvents().stream()
+ .filter(e -> e.getEventType() ==
ProvenanceEventType.ATTRIBUTES_MODIFIED)
+ .findFirst()
+ .orElse(null);
+ assertNotNull(modifyEvent, "Should have an ATTRIBUTES_MODIFIED
provenance event");
+ }
+
+ @Test
+ void testMetadataAttributePrefix() {
+ when(blobProperties.getMetadata()).thenReturn(Map.of("customKey",
"customValue"));
+
+ runner.enqueue("");
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(
+ AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND, 1);
+
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(
+
AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND).getFirst();
+ assertEquals("customValue",
flowFile.getAttribute("azure.user.metadata.customKey"));
+ assertNull(flowFile.getAttribute("customKey"),
+ "Raw key should not appear without prefix");
+ }
+
+ @Test
+ void testMultipleFlowFiles() {
+ when(blobProperties.getMetadata())
+ .thenReturn(Map.of("seq", "1"))
+ .thenReturn(Map.of("seq", "2"));
+
+ runner.enqueue("");
+ runner.enqueue("");
+ runner.run(2);
+
+ assertEquals(2, runner.getFlowFilesForRelationship(
+
AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND).size());
+ }
+
+ @Test
+ void testMetadataWithManyEntries() {
+ final Map<String, String> largeMetadata = new HashMap<>();
+ for (int i = 0; i < 20; i++) {
+ largeMetadata.put("key" + i, "value" + i);
+ }
+ when(blobProperties.getMetadata()).thenReturn(largeMetadata);
+
+ runner.enqueue("");
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(
+ AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND, 1);
+
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(
+
AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND).getFirst();
+
+ for (int i = 0; i < 20; i++) {
+ assertEquals("value" + i,
+ flowFile.getAttribute("azure.user.metadata.key" + i));
+ }
+ }
+
+ private static BlobStorageException mockBlobStorageException(BlobErrorCode
errorCode) {
+ final BlobStorageException exception =
mock(BlobStorageException.class);
+ when(exception.getErrorCode()).thenReturn(errorCode);
+ when(exception.getMessage()).thenReturn("Mocked: " + errorCode);
+ return exception;
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/TestFetchAzureBlobStorageTags.java
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/TestFetchAzureBlobStorageTags.java
new file mode 100644
index 00000000000..1c9a4a6b4ce
--- /dev/null
+++
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/TestFetchAzureBlobStorageTags.java
@@ -0,0 +1,237 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.azure.storage;
+
+import com.azure.storage.blob.BlobClient;
+import com.azure.storage.blob.BlobContainerClient;
+import com.azure.storage.blob.BlobServiceClient;
+import com.azure.storage.blob.models.BlobErrorCode;
+import com.azure.storage.blob.models.BlobStorageException;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.provenance.ProvenanceEventRecord;
+import org.apache.nifi.provenance.ProvenanceEventType;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+import static
org.apache.nifi.processors.azure.storage.utils.AzureStorageUtils.BLOB_STORAGE_CREDENTIALS_SERVICE;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TestFetchAzureBlobStorageTags {
+
+ private static final String CONTAINER_NAME = "test-container";
+ private static final String BLOB_NAME = "test-blob";
+
+ private TestRunner runner;
+ private BlobServiceClient storageClient;
+ private BlobClient blobClient;
+
+ @BeforeEach
+ void setUp() {
+ storageClient = mock(BlobServiceClient.class);
+ final BlobContainerClient containerClient =
mock(BlobContainerClient.class);
+ blobClient = mock(BlobClient.class);
+
+
when(storageClient.getBlobContainerClient(CONTAINER_NAME)).thenReturn(containerClient);
+ when(containerClient.getBlobClient(BLOB_NAME)).thenReturn(blobClient);
+
+ final FetchAzureBlobStorageTags processor = new
FetchAzureBlobStorageTags() {
+ @Override
+ protected BlobServiceClient getStorageClient(PropertyContext
context, FlowFile flowFile) {
+ return storageClient;
+ }
+
+ @Override
+ protected List<PropertyDescriptor>
getSupportedPropertyDescriptors() {
+ return super.getSupportedPropertyDescriptors().stream()
+ .filter(pd ->
!pd.equals(BLOB_STORAGE_CREDENTIALS_SERVICE))
+ .toList();
+ }
+ };
+
+ runner = TestRunners.newTestRunner(processor);
+
runner.setProperty(AbstractFetchAzureBlobStoragePropertiesProcessor.CONTAINER,
CONTAINER_NAME);
+
runner.setProperty(AbstractFetchAzureBlobStoragePropertiesProcessor.BLOB_NAME,
BLOB_NAME);
+ }
+
+ @Test
+ void testSuccessfulTagRetrieval() {
+ final Map<String, String> tags = Map.of(
+ "environment", "production",
+ "department", "engineering",
+ "cost-center", "12345"
+ );
+ when(blobClient.getTags()).thenReturn(tags);
+
+ runner.enqueue("");
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(
+ AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND, 1);
+
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(
+
AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND).getFirst();
+
+ assertEquals("production",
flowFile.getAttribute("azure.tag.environment"));
+ assertEquals("engineering",
flowFile.getAttribute("azure.tag.department"));
+ assertEquals("12345", flowFile.getAttribute("azure.tag.cost-center"));
+ }
+
+ @Test
+ void testEmptyTagsRetrieval() {
+ when(blobClient.getTags()).thenReturn(Map.of());
+
+ runner.enqueue("");
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(
+ AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND, 1);
+
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(
+
AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND).getFirst();
+
+ flowFile.getAttributes().forEach((key, value) ->
+ assertFalse(key.startsWith("azure.tag."),
+ "No tag attributes should be set when blob has no tags,
found: " + key)
+ );
+ }
+
+ @Test
+ void testBlobNotFound() {
+ BlobStorageException exception =
mockBlobStorageException(BlobErrorCode.BLOB_NOT_FOUND);
+ when(blobClient.getTags()).thenThrow(exception);
+
+ runner.enqueue("");
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(
+
AbstractFetchAzureBlobStoragePropertiesProcessor.REL_NOT_FOUND, 1);
+ }
+
+ @Test
+ void testOtherBlobStorageExceptionRoutesToFailure() {
+ BlobStorageException exception =
mockBlobStorageException(BlobErrorCode.AUTHORIZATION_FAILURE);
+ when(blobClient.getTags()).thenThrow(exception);
+
+ runner.enqueue("");
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(
+ AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FAILURE,
1);
+
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(
+
AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FAILURE).getFirst();
+ assertTrue(flowFile.isPenalized(), "FlowFile should be penalized on
failure");
+ }
+
+ @Test
+ void testContainerAndBlobNameFromFlowFileAttributes() {
+
runner.setProperty(AbstractFetchAzureBlobStoragePropertiesProcessor.CONTAINER,
+ "${azure.container}");
+
runner.setProperty(AbstractFetchAzureBlobStoragePropertiesProcessor.BLOB_NAME,
+ "${azure.blobname}");
+
+ final String dynamicContainer = "dynamic-container";
+ final String dynamicBlob = "dynamic-blob";
+
+ final BlobContainerClient dynamicContainerClient =
mock(BlobContainerClient.class);
+ final BlobClient dynamicBlobClient = mock(BlobClient.class);
+ when(storageClient.getBlobContainerClient(dynamicContainer))
+ .thenReturn(dynamicContainerClient);
+ when(dynamicContainerClient.getBlobClient(dynamicBlob))
+ .thenReturn(dynamicBlobClient);
+ when(dynamicBlobClient.getTags()).thenReturn(Map.of("region",
"us-east"));
+
+ runner.enqueue("", Map.of(
+ "azure.container", dynamicContainer,
+ "azure.blobname", dynamicBlob
+ ));
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(
+ AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND, 1);
+
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(
+
AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND).getFirst();
+ assertEquals("us-east", flowFile.getAttribute("azure.tag.region"));
+ }
+
+ @Test
+ void testProvenanceEventOnFound() {
+ when(blobClient.getTags()).thenReturn(Map.of("key", "value"));
+
+ runner.enqueue("");
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(
+ AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND, 1);
+
+ final ProvenanceEventRecord modifyEvent =
runner.getProvenanceEvents().stream()
+ .filter(e -> e.getEventType() ==
ProvenanceEventType.ATTRIBUTES_MODIFIED)
+ .findFirst()
+ .orElse(null);
+ assertNotNull(modifyEvent, "Should have an ATTRIBUTES_MODIFIED
provenance event");
+ }
+
+ @Test
+ void testTagAttributePrefix() {
+ when(blobClient.getTags()).thenReturn(Map.of("myKey", "myValue"));
+
+ runner.enqueue("");
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(
+ AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND, 1);
+
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(
+
AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND).getFirst();
+ assertEquals("myValue", flowFile.getAttribute("azure.tag.myKey"));
+ }
+
+ @Test
+ void testMultipleFlowFiles() {
+ when(blobClient.getTags())
+ .thenReturn(Map.of("batch", "1"))
+ .thenReturn(Map.of("batch", "2"));
+
+ runner.enqueue("");
+ runner.enqueue("");
+ runner.run(2);
+
+ assertEquals(2, runner.getFlowFilesForRelationship(
+
AbstractFetchAzureBlobStoragePropertiesProcessor.REL_FOUND).size());
+ }
+
+ private static BlobStorageException mockBlobStorageException(BlobErrorCode
errorCode) {
+ final BlobStorageException exception =
mock(BlobStorageException.class);
+ when(exception.getErrorCode()).thenReturn(errorCode);
+ when(exception.getMessage()).thenReturn("Mocked: " + errorCode);
+ return exception;
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/TestPutAzureBlobStorage_v12.java
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/TestPutAzureBlobStorage_v12.java
index 74f78e66645..cf7cf8a23c5 100644
---
a/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/TestPutAzureBlobStorage_v12.java
+++
b/nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/TestPutAzureBlobStorage_v12.java
@@ -16,19 +16,62 @@
*/
package org.apache.nifi.processors.azure.storage;
+import com.azure.core.http.rest.Response;
+import com.azure.core.util.Context;
+import com.azure.storage.blob.BlobClient;
+import com.azure.storage.blob.BlobContainerClient;
+import com.azure.storage.blob.BlobServiceClient;
+import com.azure.storage.blob.models.BlobProperties;
+import com.azure.storage.blob.models.BlobType;
+import com.azure.storage.blob.models.BlockBlobItem;
+import com.azure.storage.blob.options.BlobParallelUploadOptions;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.migration.ProxyServiceMigration;
import org.apache.nifi.processors.azure.AbstractAzureBlobProcessor_v12;
import org.apache.nifi.processors.azure.storage.utils.AzureStorageUtils;
+import org.apache.nifi.provenance.ProvenanceEventRecord;
+import org.apache.nifi.provenance.ProvenanceEventType;
+import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.PropertyMigrationResult;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import java.time.OffsetDateTime;
+import java.util.HashMap;
+import java.util.List;
import java.util.Map;
+import static
org.apache.nifi.processors.azure.storage.utils.AzureStorageUtils.BLOB_STORAGE_CREDENTIALS_SERVICE;
+import static
org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR_NAME_USER_METADATA;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
public class TestPutAzureBlobStorage_v12 {
+
+ private static final String CONTAINER_NAME = "test-container";
+ private static final String BLOB_NAME = "test-blob";
+ private static final String TEST_CONTENT = "test content";
+ private static final String ETAG = "test-etag";
+ private static final String BLOB_URL =
"https://account.blob.core.windows.net/test-container/test-blob";;
+
+ private TestRunner runner;
+ private BlobClient blobClient;
+ private ArgumentCaptor<BlobParallelUploadOptions> uploadOptionsCaptor;
+
@Test
void testMigration() {
TestRunner runner =
TestRunners.newTestRunner(PutAzureBlobStorage_v12.class);
@@ -43,4 +86,306 @@ public class TestPutAzureBlobStorage_v12 {
assertEquals(expectedRenamed,
propertyMigrationResult.getPropertiesRenamed());
}
+
+ @Nested
+ class PutAzureTests {
+
+ @BeforeEach
+ void setUp() {
+ final BlobServiceClient storageClient =
mock(BlobServiceClient.class);
+ final BlobContainerClient containerClient =
mock(BlobContainerClient.class);
+ blobClient = mock(BlobClient.class);
+ uploadOptionsCaptor =
ArgumentCaptor.forClass(BlobParallelUploadOptions.class);
+
+
when(storageClient.getBlobContainerClient(CONTAINER_NAME)).thenReturn(containerClient);
+
when(containerClient.getBlobClient(BLOB_NAME)).thenReturn(blobClient);
+ when(containerClient.exists()).thenReturn(true);
+ when(blobClient.getBlobUrl()).thenReturn(BLOB_URL);
+ when(blobClient.getContainerName()).thenReturn(CONTAINER_NAME);
+ when(blobClient.getBlobName()).thenReturn(BLOB_NAME);
+
+ final BlockBlobItem blockBlobItem = mock(BlockBlobItem.class);
+ when(blockBlobItem.getETag()).thenReturn(ETAG);
+
when(blockBlobItem.getLastModified()).thenReturn(OffsetDateTime.now());
+
+ final Response<BlockBlobItem> response = mock(Response.class);
+ when(response.getValue()).thenReturn(blockBlobItem);
+ when(blobClient.uploadWithResponse(
+ any(BlobParallelUploadOptions.class), isNull(),
any(Context.class)))
+ .thenReturn(response);
+
+ final BlobProperties blobProperties = mock(BlobProperties.class);
+ when(blobProperties.getBlobType()).thenReturn(BlobType.BLOCK_BLOB);
+ when(blobProperties.getETag()).thenReturn(ETAG);
+
when(blobProperties.getContentType()).thenReturn("application/octet-stream");
+
when(blobProperties.getLastModified()).thenReturn(OffsetDateTime.now());
+ when(blobProperties.getBlobSize()).thenReturn((long)
TEST_CONTENT.length());
+ when(blobProperties.getContentLanguage()).thenReturn(null);
+ when(blobClient.getProperties()).thenReturn(blobProperties);
+
+ final PutAzureBlobStorage_v12 processor = new
PutAzureBlobStorage_v12() {
+ @Override
+ protected BlobServiceClient getStorageClient(PropertyContext
context, FlowFile flowFile) {
+ return storageClient;
+ }
+
+ @Override
+ protected List<PropertyDescriptor>
getSupportedPropertyDescriptors() {
+ return super.getSupportedPropertyDescriptors().stream()
+ .filter(pd ->
!pd.equals(BLOB_STORAGE_CREDENTIALS_SERVICE))
+ .toList();
+ }
+ };
+
+ runner = TestRunners.newTestRunner(processor);
+ runner.setProperty(AbstractAzureBlobProcessor_v12.BLOB_NAME,
BLOB_NAME);
+ runner.setProperty("Container Name", CONTAINER_NAME);
+ }
+
+ @Test
+ void testTagsFromFlowFileAttributesWithPrefix() {
+ runner.setProperty(PutAzureBlobStorage_v12.BLOB_TAG_PREFIX,
"azure.tag.");
+
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put("azure.tag.environment", "production");
+ attributes.put("azure.tag.team", "engineering");
+ attributes.put("other.attribute", "should-be-ignored");
+
+ runner.enqueue(TEST_CONTENT, attributes);
+ runner.run();
+
+
runner.assertAllFlowFilesTransferred(AbstractAzureBlobProcessor_v12.REL_SUCCESS,
1);
+
+ verify(blobClient).uploadWithResponse(
+ uploadOptionsCaptor.capture(), isNull(),
any(Context.class));
+ final Map<String, String> tags =
uploadOptionsCaptor.getValue().getTags();
+
+ assertNotNull(tags, "Tags should not be null when tag prefix is
set");
+ assertEquals(2, tags.size());
+ assertEquals("production", tags.get("azure.tag.environment"));
+ assertEquals("engineering", tags.get("azure.tag.team"));
+ }
+
+ @Test
+ void testTagsFromFlowFileAttributesWithPrefixRemoval() {
+ runner.setProperty(PutAzureBlobStorage_v12.BLOB_TAG_PREFIX,
"azure.tag.");
+ runner.setProperty(PutAzureBlobStorage_v12.REMOVE_TAG_PREFIX,
"true");
+
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put("azure.tag.environment", "staging");
+ attributes.put("azure.tag.department", "finance");
+ attributes.put("unrelated.key", "unrelated-value");
+
+ runner.enqueue(TEST_CONTENT, attributes);
+ runner.run();
+
+
runner.assertAllFlowFilesTransferred(AbstractAzureBlobProcessor_v12.REL_SUCCESS,
1);
+
+ verify(blobClient).uploadWithResponse(
+ uploadOptionsCaptor.capture(), isNull(),
any(Context.class));
+ final Map<String, String> tags =
uploadOptionsCaptor.getValue().getTags();
+
+ assertNotNull(tags);
+ assertEquals(2, tags.size());
+ assertEquals("staging", tags.get("environment"));
+ assertEquals("finance", tags.get("department"));
+ }
+
+ @Test
+ void testNoTagsWhenPrefixNotSet() {
+ runner.enqueue(TEST_CONTENT, Map.of("azure.tag.something",
"value"));
+ runner.run();
+
+
runner.assertAllFlowFilesTransferred(AbstractAzureBlobProcessor_v12.REL_SUCCESS,
1);
+
+ verify(blobClient).uploadWithResponse(
+ uploadOptionsCaptor.capture(), isNull(),
any(Context.class));
+ assertNull(uploadOptionsCaptor.getValue().getTags(),
+ "Tags should be null when tag prefix property is not set");
+ }
+
+ @Test
+ void testNoMatchingTagAttributes() {
+ runner.setProperty(PutAzureBlobStorage_v12.BLOB_TAG_PREFIX,
"azure.tag.");
+
+ runner.enqueue(TEST_CONTENT, Map.of("other.key", "other-value"));
+ runner.run();
+
+
runner.assertAllFlowFilesTransferred(AbstractAzureBlobProcessor_v12.REL_SUCCESS,
1);
+
+ verify(blobClient).uploadWithResponse(
+ uploadOptionsCaptor.capture(), isNull(),
any(Context.class));
+ final Map<String, String> tags =
uploadOptionsCaptor.getValue().getTags();
+
+ assertNotNull(tags);
+ assertTrue(tags.isEmpty(),
+ "Tags should be empty when no attributes match the
prefix");
+ }
+
+ @Test
+ void testTagPrefixWithExpressionLanguage() {
+ runner.setProperty(PutAzureBlobStorage_v12.BLOB_TAG_PREFIX,
"${tag.prefix}");
+
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put("tag.prefix", "custom.");
+ attributes.put("custom.region", "us-east-1");
+ attributes.put("custom.tier", "standard");
+
+ runner.enqueue(TEST_CONTENT, attributes);
+ runner.run();
+
+
runner.assertAllFlowFilesTransferred(AbstractAzureBlobProcessor_v12.REL_SUCCESS,
1);
+
+ verify(blobClient).uploadWithResponse(
+ uploadOptionsCaptor.capture(), isNull(),
any(Context.class));
+ final Map<String, String> tags =
uploadOptionsCaptor.getValue().getTags();
+
+ assertNotNull(tags);
+ assertEquals(2, tags.size());
+ assertEquals("us-east-1", tags.get("custom.region"));
+ assertEquals("standard", tags.get("custom.tier"));
+ }
+
+ @Test
+ void testRemoveTagPrefixDefaultsToFalse() {
+ runner.setProperty(PutAzureBlobStorage_v12.BLOB_TAG_PREFIX,
"pfx.");
+
+ runner.enqueue(TEST_CONTENT, Map.of("pfx.key", "val"));
+ runner.run();
+
+
runner.assertAllFlowFilesTransferred(AbstractAzureBlobProcessor_v12.REL_SUCCESS,
1);
+
+ verify(blobClient).uploadWithResponse(
+ uploadOptionsCaptor.capture(), isNull(),
any(Context.class));
+ final Map<String, String> tags =
uploadOptionsCaptor.getValue().getTags();
+
+ assertNotNull(tags);
+ assertTrue(tags.containsKey("pfx.key"),
+ "Full attribute name should be the tag key when prefix
removal is off");
+ assertFalse(tags.containsKey("key"),
+ "Stripped key should not appear when prefix removal is
off");
+ }
+
+ @Test
+ void testUserMetadataFromDynamicProperties() {
+ runner.setProperty("x-custom-header", "header-value");
+ runner.setProperty("department", "engineering");
+
+ runner.enqueue(TEST_CONTENT);
+ runner.run();
+
+
runner.assertAllFlowFilesTransferred(AbstractAzureBlobProcessor_v12.REL_SUCCESS,
1);
+
+ verify(blobClient).uploadWithResponse(
+ uploadOptionsCaptor.capture(), isNull(),
any(Context.class));
+ final Map<String, String> metadata =
+ uploadOptionsCaptor.getValue().getMetadata();
+
+ assertNotNull(metadata);
+ assertEquals(2, metadata.size());
+ assertEquals("header-value", metadata.get("x-custom-header"));
+ assertEquals("engineering", metadata.get("department"));
+ }
+
+ @Test
+ void testUserMetadataWithExpressionLanguage() {
+ runner.setProperty("source-system", "${system.name}");
+
+ runner.enqueue(TEST_CONTENT, Map.of("system.name", "crm-export"));
+ runner.run();
+
+
runner.assertAllFlowFilesTransferred(AbstractAzureBlobProcessor_v12.REL_SUCCESS,
1);
+
+ verify(blobClient).uploadWithResponse(
+ uploadOptionsCaptor.capture(), isNull(),
any(Context.class));
+ final Map<String, String> metadata =
+ uploadOptionsCaptor.getValue().getMetadata();
+
+ assertNotNull(metadata);
+ assertEquals("crm-export", metadata.get("source-system"));
+ }
+
+ @Test
+ void testNoUserMetadataWhenNoDynamicProperties() {
+ runner.enqueue(TEST_CONTENT);
+ runner.run();
+
+
runner.assertAllFlowFilesTransferred(AbstractAzureBlobProcessor_v12.REL_SUCCESS,
1);
+
+ verify(blobClient).uploadWithResponse(
+ uploadOptionsCaptor.capture(), isNull(),
any(Context.class));
+ assertNull(uploadOptionsCaptor.getValue().getMetadata(),
+ "Metadata should not be set when no dynamic properties
exist");
+ }
+
+ @Test
+ void testUserMetadataAttributeOnSuccessFlowFile() {
+ runner.setProperty("project", "alpha");
+ runner.setProperty("owner", "team-a");
+
+ runner.enqueue(TEST_CONTENT);
+ runner.run();
+
+
runner.assertAllFlowFilesTransferred(AbstractAzureBlobProcessor_v12.REL_SUCCESS,
1);
+
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(
+ AbstractAzureBlobProcessor_v12.REL_SUCCESS).getFirst();
+ final String attr = flowFile.getAttribute(ATTR_NAME_USER_METADATA);
+
+ assertNotNull(attr, "User metadata attribute should be present");
+ assertTrue(attr.contains("project=alpha"));
+ assertTrue(attr.contains("owner=team-a"));
+ }
+
+ @Test
+ void testNoUserMetadataAttributeWhenNoDynamicProperties() {
+ runner.enqueue(TEST_CONTENT);
+ runner.run();
+
+
runner.assertAllFlowFilesTransferred(AbstractAzureBlobProcessor_v12.REL_SUCCESS,
1);
+
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(
+ AbstractAzureBlobProcessor_v12.REL_SUCCESS).getFirst();
+ assertNull(flowFile.getAttribute(ATTR_NAME_USER_METADATA),
+ "User metadata attribute should be absent when no dynamic
properties exist");
+ }
+
+ @Test
+ void testBothTagsAndMetadataSet() {
+ runner.setProperty(PutAzureBlobStorage_v12.BLOB_TAG_PREFIX,
"tag.");
+ runner.setProperty(PutAzureBlobStorage_v12.REMOVE_TAG_PREFIX,
"true");
+ runner.setProperty("meta-key", "meta-value");
+
+ runner.enqueue(TEST_CONTENT, Map.of("tag.env", "prod"));
+ runner.run();
+
+
runner.assertAllFlowFilesTransferred(AbstractAzureBlobProcessor_v12.REL_SUCCESS,
1);
+
+ verify(blobClient).uploadWithResponse(
+ uploadOptionsCaptor.capture(), isNull(),
any(Context.class));
+ final BlobParallelUploadOptions options =
uploadOptionsCaptor.getValue();
+
+ assertEquals("prod", options.getTags().get("env"));
+ assertEquals("meta-value", options.getMetadata().get("meta-key"));
+
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(
+ AbstractAzureBlobProcessor_v12.REL_SUCCESS).getFirst();
+ assertNotNull(flowFile.getAttribute(ATTR_NAME_USER_METADATA));
+ }
+
+ @Test
+ void testProvenanceEventOnSuccess() {
+ runner.enqueue(TEST_CONTENT);
+ runner.run();
+
+
runner.assertAllFlowFilesTransferred(AbstractAzureBlobProcessor_v12.REL_SUCCESS,
1);
+
+ final ProvenanceEventRecord sendEvent =
runner.getProvenanceEvents().stream()
+ .filter(e -> e.getEventType() == ProvenanceEventType.SEND)
+ .findFirst()
+ .orElse(null);
+ assertNotNull(sendEvent, "Should have a SEND provenance event");
+ assertTrue(sendEvent.getTransitUri().contains(BLOB_NAME));
+ }
+ }
}
