This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits 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 74c07276a71 NIFI-15969 Fixed PutS3Object multipart upload data
corruption for concurrent FlowFiles with same S3 key
74c07276a71 is described below
commit 74c07276a71bae160da571058eebd5c4196f7c09
Author: Rakesh Kumar Singh <[email protected]>
AuthorDate: Mon May 25 13:22:07 2026 +0530
NIFI-15969 Fixed PutS3Object multipart upload data corruption for
concurrent FlowFiles with same S3 key
Previously the multipart upload state was tracked using only the processor
identifier,
bucket name, and object key. When two FlowFiles with the same name were
uploaded
concurrently to the same bucket, they shared the same state tracking key,
causing
parts from different uploads to be interleaved and resulting in a corrupt
S3 object.
Included the FlowFile UUID in the state tracking key so each FlowFile
maintains
its own independent multipart upload state. Retries of the same FlowFile
retain
the same UUID and continue to benefit from state resumption. A FlowFile
with a
new UUID starts a fresh upload rather than inheriting stale state.
This closes #11279.
Signed-off-by: Peter Turcsanyi <[email protected]>
---
.../apache/nifi/processors/aws/s3/PutS3Object.java | 6 +++++-
.../nifi/processors/aws/s3/TestPutS3Object.java | 20 ++++++++++++++++++++
2 files changed, 25 insertions(+), 1 deletion(-)
diff --git
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/s3/PutS3Object.java
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/s3/PutS3Object.java
index 9278f218ceb..9889d618a55 100644
---
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/s3/PutS3Object.java
+++
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/s3/PutS3Object.java
@@ -394,6 +394,10 @@ public class PutS3Object extends AbstractS3Processor {
return new File(this.tempDirMultipart + File.separator +
getIdentifier());
}
+ protected String buildCacheKey(final String bucket, final String key,
final String uuid) {
+ return getIdentifier() + "/" + bucket + "/" + key + "/" + uuid;
+ }
+
protected boolean localUploadExistsInS3(final S3Client client, final
String bucket, final MultipartState localState) {
final ListMultipartUploadsRequest listRequest =
ListMultipartUploadsRequest.builder()
.bucket(bucket)
@@ -546,7 +550,7 @@ public class PutS3Object extends AbstractS3Processor {
final String bucket =
context.getProperty(BUCKET_WITH_DEFAULT_VALUE).evaluateAttributeExpressions(flowFile).getValue();
final String key =
context.getProperty(KEY).evaluateAttributeExpressions(flowFile).getValue();
- final String cacheKey = getIdentifier() + "/" + bucket + "/" + key;
+ final String cacheKey = buildCacheKey(bucket, key,
flowFile.getAttribute(CoreAttributes.UUID.key()));
final Map<String, String> attributes = new HashMap<>();
final String ffFilename =
flowFile.getAttributes().get(CoreAttributes.FILENAME.key());
diff --git
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/s3/TestPutS3Object.java
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/s3/TestPutS3Object.java
index 9d75bf3b098..a031a3afa3e 100644
---
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/s3/TestPutS3Object.java
+++
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/s3/TestPutS3Object.java
@@ -51,11 +51,13 @@ import java.net.URLEncoder;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
import static java.nio.charset.StandardCharsets.UTF_8;
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.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.mockito.ArgumentMatchers.anyMap;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
@@ -291,4 +293,22 @@ public class TestPutS3Object {
expectedRenamed.forEach((key, value) -> assertEquals(value,
propertyMigrationResult.getPropertiesRenamed().get(key)));
}
+
+ @Test
+ void testBuildCacheKeyIncludesFlowFileUUID() {
+ final String bucket = "test-bucket";
+ final String key = "data/file.csv";
+ final String uuid1 = UUID.randomUUID().toString();
+ final String uuid2 = UUID.randomUUID().toString();
+
+ final String cacheKey1 = putS3Object.buildCacheKey(bucket, key, uuid1);
+ final String cacheKey2 = putS3Object.buildCacheKey(bucket, key, uuid2);
+ final String cacheKey1Retry = putS3Object.buildCacheKey(bucket, key,
uuid1);
+
+ assertNotEquals(cacheKey1, cacheKey2,
+ "FlowFiles with different UUIDs sharing the same bucket and
key must use distinct cache keys to prevent multipart state collision");
+ assertEquals(cacheKey1, cacheKey1Retry,
+ "Same FlowFile UUID must produce the same cache key so
interrupted uploads can be resumed");
+ }
+
}
