Skip to content

NIFI-15969 Fixed PutS3Object multipart upload data corruption for concurrent FlowFiles with same S3 key #11279

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
rakesh-rsky wants to merge 1 commit into from
+25 −1
Closed

NIFI-15969 Fixed PutS3Object multipart upload data corruption for concurrent FlowFiles with same S3 key #11279

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion ...ndle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/s3/PutS3Object.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -394,6 +394,10 @@ protected File getPersistenceFile() {
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)
Expand Down Expand Up @@ -546,7 +550,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session

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());
Expand Down
20 changes: 20 additions & 0 deletions .../nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/s3/TestPutS3Object.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,11 +51,13 @@
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;
Expand Down Expand Up @@ -291,4 +293,22 @@ void testMigration() {

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");
}

}
Loading