@Override
public SetObjectTaggingResult setObjectTagging(SetObjectTaggingRequest setObjectTaggingRequest, AmazonS3 s3Client)
{
    MockS3Object mockS3Object =
        getMockS3Object(setObjectTaggingRequest.getBucketName(), setObjectTaggingRequest.getKey(), setObjectTaggingRequest.getVersionId());

    if (setObjectTaggingRequest.getTagging() != null)
    {
        mockS3Object.setTags(setObjectTaggingRequest.getTagging().getTagSet());
    }
    else
    {
        mockS3Object.setTags(null);
    }

    return new SetObjectTaggingResult();
}
 

@Test
public void testTagObjectVersion() throws IOException {
    final String tagKey = "k";
    final String tagVal = "v";
    runner.setProperty(TagS3Object.REGION, "us-west-2");
    runner.setProperty(TagS3Object.BUCKET, "test-bucket");
    runner.setProperty(TagS3Object.VERSION_ID, "test-version");
    runner.setProperty(TagS3Object.TAG_KEY, tagKey);
    runner.setProperty(TagS3Object.TAG_VALUE, tagVal);
    runner.setProperty(TagS3Object.APPEND_TAG, "false");
    final Map<String, String> attrs = new HashMap<>();
    attrs.put("filename", "object-key");
    runner.enqueue(new byte[0], attrs);

    runner.run(1);

    runner.assertAllFlowFilesTransferred(TagS3Object.REL_SUCCESS, 1);
    ArgumentCaptor<SetObjectTaggingRequest> captureRequest = ArgumentCaptor.forClass(SetObjectTaggingRequest.class);
    Mockito.verify(mockS3Client, Mockito.times(1)).setObjectTagging(captureRequest.capture());
    SetObjectTaggingRequest request = captureRequest.getValue();
    assertEquals("test-bucket", request.getBucketName());
    assertEquals("object-key", request.getKey());
    assertEquals("test-version", request.getVersionId());
    assertTrue("Expected tag not found in request", request.getTagging().getTagSet().contains(new Tag(tagKey, tagVal)));
}
 

@Test
public void testTagObjectS3Exception() {
    //set up existing tags on S3 object
    Tag currentTag = new Tag("ck", "cv");
    mockGetExistingTags(currentTag);

    final String tagKey = "nk";
    final String tagVal = "nv";
    runner.setProperty(TagS3Object.REGION, "us-west-2");
    runner.setProperty(TagS3Object.BUCKET, "test-bucket");
    runner.setProperty(TagS3Object.TAG_KEY, tagKey);
    runner.setProperty(TagS3Object.TAG_VALUE, tagVal);
    final Map<String, String> attrs = new HashMap<>();
    attrs.put("filename", "delete-key");
    runner.enqueue(new byte[0], attrs);
    Mockito.doThrow(new AmazonS3Exception("TagFailure")).when(mockS3Client).setObjectTagging(Mockito.any());

    runner.run(1);

    runner.assertAllFlowFilesTransferred(DeleteS3Object.REL_FAILURE, 1);
    ArgumentCaptor<SetObjectTaggingRequest> captureRequest = ArgumentCaptor.forClass(SetObjectTaggingRequest.class);
}
 

private SetObjectTaggingRequest tagAsDeleted(final String key) {
  return new SetObjectTaggingRequest(
      getConfiguredBucket(),
      key,
      new ObjectTagging(singletonList(DELETED_TAG))
  );
}
 

private SetObjectTaggingRequest untagAsDeleted(final String key) {
  return new SetObjectTaggingRequest(
      getConfiguredBucket(),
      key,
      new ObjectTagging(emptyList())
  );
}
 

private void changeExistingObject(
  Record record,
  ELVars variables,
  String bucket,
  String objectPath
) throws OnRecordErrorException {
  // Tag application
  if(!config.taskConfig.tags.isEmpty()) {
    List<Tag> newTags = new ArrayList<>();

    // Evaluate each tag separately
    for (Map.Entry<String, String> entry : config.taskConfig.tags.entrySet()) {
      newTags.add(new Tag(
        evaluate(record, "tags", variables, entry.getKey()),
        evaluate(record, "tags", variables, entry.getValue())
      ));
    }

    // Apply all tags at once
    config.s3Config.getS3Client().setObjectTagging(new SetObjectTaggingRequest(
      bucket,
      objectPath,
      new ObjectTagging(newTags)
    ));

    Events.FILE_CHANGED.create(getContext())
      .with("object_key", objectPath)
      .createAndSend();
  }
}
 

@Test
public void testTagObjectSimple() throws IOException {
    final String tagKey = "k";
    final String tagVal = "v";
    runner.setProperty(TagS3Object.REGION, "us-west-2");
    runner.setProperty(TagS3Object.BUCKET, "test-bucket");
    runner.setProperty(TagS3Object.TAG_KEY, tagKey);
    runner.setProperty(TagS3Object.TAG_VALUE, tagVal);
    runner.setProperty(TagS3Object.APPEND_TAG, "false");
    final Map<String, String> attrs = new HashMap<>();
    attrs.put("filename", "object-key");
    runner.enqueue(new byte[0], attrs);

    runner.run(1);

    runner.assertAllFlowFilesTransferred(TagS3Object.REL_SUCCESS, 1);
    ArgumentCaptor<SetObjectTaggingRequest> captureRequest = ArgumentCaptor.forClass(SetObjectTaggingRequest.class);
    Mockito.verify(mockS3Client, Mockito.times(1)).setObjectTagging(captureRequest.capture());
    SetObjectTaggingRequest request = captureRequest.getValue();
    assertEquals("test-bucket", request.getBucketName());
    assertEquals("object-key", request.getKey());
    assertNull("test-version", request.getVersionId());
    assertTrue("Expected tag not found in request", request.getTagging().getTagSet().contains(new Tag(tagKey, tagVal)));

    List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(ListS3.REL_SUCCESS);
    MockFlowFile ff0 = flowFiles.get(0);
    ff0.assertAttributeEquals("s3.tag."+tagKey, tagVal);
}
 

@Test
public void testTagObjectAppendToExistingTags() throws IOException {
    //set up existing tags on S3 object
    Tag currentTag = new Tag("ck", "cv");
    mockGetExistingTags(currentTag);

    final String tagKey = "nk";
    final String tagVal = "nv";
    runner.setProperty(TagS3Object.REGION, "us-west-2");
    runner.setProperty(TagS3Object.BUCKET, "test-bucket");
    runner.setProperty(TagS3Object.TAG_KEY, tagKey);
    runner.setProperty(TagS3Object.TAG_VALUE, tagVal);
    final Map<String, String> attrs = new HashMap<>();
    attrs.put("filename", "object-key");
    attrs.put("s3.tag."+currentTag.getKey(), currentTag.getValue());
    runner.enqueue(new byte[0], attrs);

    runner.run(1);

    runner.assertAllFlowFilesTransferred(TagS3Object.REL_SUCCESS, 1);
    ArgumentCaptor<SetObjectTaggingRequest> captureRequest = ArgumentCaptor.forClass(SetObjectTaggingRequest.class);
    Mockito.verify(mockS3Client, Mockito.times(1)).setObjectTagging(captureRequest.capture());
    SetObjectTaggingRequest request = captureRequest.getValue();
    assertEquals("test-bucket", request.getBucketName());
    assertEquals("object-key", request.getKey());
    assertTrue("New tag not found in request", request.getTagging().getTagSet().contains(new Tag(tagKey, tagVal)));
    assertTrue("Existing tag not found in request", request.getTagging().getTagSet().contains(currentTag));

    List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(ListS3.REL_SUCCESS);
    MockFlowFile ff0 = flowFiles.get(0);
    ff0.assertAttributeEquals("s3.tag."+tagKey, tagVal);
    ff0.assertAttributeEquals("s3.tag."+currentTag.getKey(), currentTag.getValue());
}
 

@Test
public void testTagObjectAppendUpdatesExistingTagValue() throws IOException {
    //set up existing tags on S3 object
    Tag currentTag1 = new Tag("ck", "cv");
    Tag currentTag2 = new Tag("nk", "ov");
    mockGetExistingTags(currentTag1, currentTag2);

    final String tagKey = "nk";
    final String tagVal = "nv";
    runner.setProperty(TagS3Object.REGION, "us-west-2");
    runner.setProperty(TagS3Object.BUCKET, "test-bucket");
    runner.setProperty(TagS3Object.TAG_KEY, tagKey);
    runner.setProperty(TagS3Object.TAG_VALUE, tagVal);
    final Map<String, String> attrs = new HashMap<>();
    attrs.put("filename", "object-key");
    runner.enqueue(new byte[0], attrs);

    runner.run(1);

    runner.assertAllFlowFilesTransferred(TagS3Object.REL_SUCCESS, 1);
    ArgumentCaptor<SetObjectTaggingRequest> captureRequest = ArgumentCaptor.forClass(SetObjectTaggingRequest.class);
    Mockito.verify(mockS3Client, Mockito.times(1)).setObjectTagging(captureRequest.capture());
    SetObjectTaggingRequest request = captureRequest.getValue();
    assertEquals("test-bucket", request.getBucketName());
    assertEquals("object-key", request.getKey());
    assertTrue("New tag not found in request", request.getTagging().getTagSet().contains(new Tag(tagKey, tagVal)));
    assertTrue("Existing tag not found in request", request.getTagging().getTagSet().contains(currentTag1));
    assertFalse("Existing tag should be excluded from request", request.getTagging().getTagSet().contains(currentTag2));
}
 

@Test
public void testTagObjectReplacesExistingTags() throws IOException {
    //set up existing tags on S3 object
    Tag currentTag = new Tag("ck", "cv");
    mockGetExistingTags(currentTag);

    final String tagKey = "nk";
    final String tagVal = "nv";
    runner.setProperty(TagS3Object.REGION, "us-west-2");
    runner.setProperty(TagS3Object.BUCKET, "test-bucket");
    runner.setProperty(TagS3Object.TAG_KEY, tagKey);
    runner.setProperty(TagS3Object.TAG_VALUE, tagVal);
    runner.setProperty(TagS3Object.APPEND_TAG, "false");
    final Map<String, String> attrs = new HashMap<>();
    attrs.put("filename", "object-key");
    attrs.put("s3.tag."+currentTag.getKey(), currentTag.getValue());
    runner.enqueue(new byte[0], attrs);

    runner.run(1);

    runner.assertAllFlowFilesTransferred(TagS3Object.REL_SUCCESS, 1);
    ArgumentCaptor<SetObjectTaggingRequest> captureRequest = ArgumentCaptor.forClass(SetObjectTaggingRequest.class);
    Mockito.verify(mockS3Client, Mockito.times(1)).setObjectTagging(captureRequest.capture());
    SetObjectTaggingRequest request = captureRequest.getValue();
    assertEquals("test-bucket", request.getBucketName());
    assertEquals("object-key", request.getKey());
    assertTrue("New tag not found in request", request.getTagging().getTagSet().contains(new Tag(tagKey, tagVal)));
    assertFalse("Existing tag should be excluded from request", request.getTagging().getTagSet().contains(currentTag));

    List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(ListS3.REL_SUCCESS);
    MockFlowFile ff0 = flowFiles.get(0);
    ff0.assertAttributeEquals("s3.tag."+tagKey, tagVal);
    ff0.assertAttributeNotExists("s3.tag."+currentTag.getKey());
}
 

@Override
@Guarded(by = STARTED)
public boolean delete(final BlobId blobId, String reason) {
  checkNotNull(blobId);

  final S3Blob blob = liveBlobs.getUnchecked(blobId);

  Lock lock = blob.lock();
  try {
    log.debug("Soft deleting blob {}", blobId);

    S3BlobAttributes blobAttributes = new S3BlobAttributes(s3, getConfiguredBucket(), attributePath(blobId).toString());

    boolean loaded = blobAttributes.load();
    if (!loaded) {
      // This could happen under some concurrent situations (two threads try to delete the same blob)
      // but it can also occur if the deleted index refers to a manually-deleted blob.
      log.warn("Attempt to mark-for-delete non-existent blob {}", blobId);
      return false;
    }
    else if (blobAttributes.isDeleted()) {
      log.debug("Attempt to delete already-deleted blob {}", blobId);
      return false;
    }

    blobAttributes.setDeleted(true);
    blobAttributes.setDeletedReason(reason);
    blobAttributes.store();

    // set "deleted=true" tag on the object, let S3 take care of deleting the blob after it expires
    s3.setObjectTagging(
        new SetObjectTaggingRequest(
            getConfiguredBucket(),
            contentPath(blobId),
            new ObjectTagging(Arrays.asList(DELETED_TAG))
        )
    );
    blob.markStale();

    return true;
  }
  catch (Exception e) {
    throw new BlobStoreException(e, blobId);
  }
  finally {
    lock.unlock();
  }
}
 

@Override
public SetObjectTaggingResult setObjectTagging(SetObjectTaggingRequest setObjectTaggingRequest, AmazonS3 s3Client)
{
    return s3Client.setObjectTagging(setObjectTaggingRequest);
}
 

private void tagVersionsHelper(final S3FileTransferRequestParamsDto s3FileTransferRequestParamsDto,
    final S3FileTransferRequestParamsDto s3ObjectTaggerParamsDto, final List<S3VersionSummary> s3VersionSummaries, final Tag tag)
{
    // Initialize an S3 version for the error message in the catch block.
    S3VersionSummary currentS3VersionSummary = s3VersionSummaries.get(0);

    // Amazon S3 client to access S3 objects.
    AmazonS3Client s3Client = null;

    // Amazon S3 client for S3 object tagging.
    AmazonS3Client s3ObjectTaggerClient = null;

    try
    {
        // Create an S3 client to access S3 objects.
        s3Client = getAmazonS3(s3FileTransferRequestParamsDto);

        // Create an S3 client for S3 object tagging.
        s3ObjectTaggerClient = getAmazonS3(s3ObjectTaggerParamsDto);

        // Create a get object tagging request.
        GetObjectTaggingRequest getObjectTaggingRequest = new GetObjectTaggingRequest(s3FileTransferRequestParamsDto.getS3BucketName(), null, null);

        // Create a set object tagging request.
        SetObjectTaggingRequest setObjectTaggingRequest = new SetObjectTaggingRequest(s3FileTransferRequestParamsDto.getS3BucketName(), null, null, null);

        for (S3VersionSummary s3VersionSummary : s3VersionSummaries)
        {
            // Set the current S3 version summary.
            currentS3VersionSummary = s3VersionSummary;

            // Retrieve the current tagging information for the S3 version.
            getObjectTaggingRequest.setKey(s3VersionSummary.getKey());
            getObjectTaggingRequest.setVersionId(s3VersionSummary.getVersionId());
            GetObjectTaggingResult getObjectTaggingResult = s3Operations.getObjectTagging(getObjectTaggingRequest, s3Client);

            // Update the list of tags to include the specified S3 object tag.
            List<Tag> updatedTags = new ArrayList<>();
            updatedTags.add(tag);
            if (CollectionUtils.isNotEmpty(getObjectTaggingResult.getTagSet()))
            {
                for (Tag currentTag : getObjectTaggingResult.getTagSet())
                {
                    if (!StringUtils.equals(tag.getKey(), currentTag.getKey()))
                    {
                        updatedTags.add(currentTag);
                    }
                }
            }

            // Update tagging information for the S3 version.
            setObjectTaggingRequest.setKey(s3VersionSummary.getKey());
            setObjectTaggingRequest.setVersionId(s3VersionSummary.getVersionId());
            setObjectTaggingRequest.setTagging(new ObjectTagging(updatedTags));
            s3Operations.setObjectTagging(setObjectTaggingRequest, s3ObjectTaggerClient);
        }
    }
    catch (Exception e)
    {
        throw new IllegalStateException(String
            .format("Failed to tag S3 object with \"%s\" key and \"%s\" version id in \"%s\" bucket. Reason: %s", currentS3VersionSummary.getKey(),
                currentS3VersionSummary.getVersionId(), s3FileTransferRequestParamsDto.getS3BucketName(), e.getMessage()), e);
    }
    finally
    {
        if (s3Client != null)
        {
            s3Client.shutdown();
        }

        if (s3ObjectTaggerClient != null)
        {
            s3ObjectTaggerClient.shutdown();
        }
    }
}
 

private void runTagObjectsTest()
{
    // Create an S3 file transfer request parameters DTO to access S3 objects.
    S3FileTransferRequestParamsDto s3FileTransferRequestParamsDto = new S3FileTransferRequestParamsDto();
    s3FileTransferRequestParamsDto.setS3BucketName(S3_BUCKET_NAME);

    // Create an S3 file transfer request parameters DTO to tag S3 objects.
    S3FileTransferRequestParamsDto s3ObjectTaggerParamsDto = new S3FileTransferRequestParamsDto();
    s3ObjectTaggerParamsDto.setAwsAccessKeyId(AWS_ASSUMED_ROLE_ACCESS_KEY);
    s3ObjectTaggerParamsDto.setAwsSecretKey(AWS_ASSUMED_ROLE_SECRET_KEY);
    s3ObjectTaggerParamsDto.setSessionToken(AWS_ASSUMED_ROLE_SESSION_TOKEN);

    // Create an S3 object summary.
    S3ObjectSummary s3ObjectSummary = new S3ObjectSummary();
    s3ObjectSummary.setKey(S3_KEY);

    // Create an S3 object tag.
    Tag tag = new Tag(S3_OBJECT_TAG_KEY, S3_OBJECT_TAG_VALUE);

    // Create a retry policy.
    RetryPolicy retryPolicy =
        new RetryPolicy(PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION, PredefinedRetryPolicies.DEFAULT_BACKOFF_STRATEGY, INTEGER_VALUE, true);

    // Create a get object tagging result.
    GetObjectTaggingResult getObjectTaggingResult = new GetObjectTaggingResult(null);

    // Create a set object tagging result.
    SetObjectTaggingResult setObjectTaggingResult = new SetObjectTaggingResult();

    // Mock the external calls.
    when(retryPolicyFactory.getRetryPolicy()).thenReturn(retryPolicy);
    when(s3Operations.getObjectTagging(any(GetObjectTaggingRequest.class), any(AmazonS3Client.class))).thenReturn(getObjectTaggingResult);
    when(s3Operations.setObjectTagging(any(SetObjectTaggingRequest.class), any(AmazonS3Client.class))).thenReturn(setObjectTaggingResult);

    // Call the method under test.
    s3DaoImpl.tagObjects(s3FileTransferRequestParamsDto, s3ObjectTaggerParamsDto, Collections.singletonList(s3ObjectSummary), tag);

    // Verify the external calls.
    verify(retryPolicyFactory, times(2)).getRetryPolicy();
    verify(s3Operations).getObjectTagging(any(GetObjectTaggingRequest.class), any(AmazonS3Client.class));
    verify(s3Operations).setObjectTagging(any(SetObjectTaggingRequest.class), any(AmazonS3Client.class));
    verifyNoMoreInteractionsHelper();
}
 

private void runTagVersionsTest()
{
    // Create an S3 file transfer request parameters DTO to access S3 objects.
    S3FileTransferRequestParamsDto s3FileTransferRequestParamsDto = new S3FileTransferRequestParamsDto();
    s3FileTransferRequestParamsDto.setS3BucketName(S3_BUCKET_NAME);

    // Create an S3 file transfer request parameters DTO to tag S3 objects.
    S3FileTransferRequestParamsDto s3ObjectTaggerParamsDto = new S3FileTransferRequestParamsDto();
    s3ObjectTaggerParamsDto.setAwsAccessKeyId(AWS_ASSUMED_ROLE_ACCESS_KEY);
    s3ObjectTaggerParamsDto.setAwsSecretKey(AWS_ASSUMED_ROLE_SECRET_KEY);
    s3ObjectTaggerParamsDto.setSessionToken(AWS_ASSUMED_ROLE_SESSION_TOKEN);

    // Create an S3 version summary.
    S3VersionSummary s3VersionSummary = new S3VersionSummary();
    s3VersionSummary.setKey(S3_KEY);
    s3VersionSummary.setVersionId(S3_VERSION_ID);

    // Create an S3 object tag.
    Tag tag = new Tag(S3_OBJECT_TAG_KEY, S3_OBJECT_TAG_VALUE);

    // Create a retry policy.
    RetryPolicy retryPolicy =
        new RetryPolicy(PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION, PredefinedRetryPolicies.DEFAULT_BACKOFF_STRATEGY, INTEGER_VALUE, true);

    // Create a get object tagging result.
    GetObjectTaggingResult getObjectTaggingResult = new GetObjectTaggingResult(null);

    // Create a set object tagging result.
    SetObjectTaggingResult setObjectTaggingResult = new SetObjectTaggingResult();

    // Mock the external calls.
    when(retryPolicyFactory.getRetryPolicy()).thenReturn(retryPolicy);
    when(s3Operations.getObjectTagging(any(GetObjectTaggingRequest.class), any(AmazonS3Client.class))).thenReturn(getObjectTaggingResult);
    when(s3Operations.setObjectTagging(any(SetObjectTaggingRequest.class), any(AmazonS3Client.class))).thenReturn(setObjectTaggingResult);

    // Call the method under test.
    s3DaoImpl.tagVersions(s3FileTransferRequestParamsDto, s3ObjectTaggerParamsDto, Collections.singletonList(s3VersionSummary), tag);

    // Verify the external calls.
    verify(retryPolicyFactory, times(2)).getRetryPolicy();
    verify(s3Operations).getObjectTagging(any(GetObjectTaggingRequest.class), any(AmazonS3Client.class));
    verify(s3Operations).setObjectTagging(any(SetObjectTaggingRequest.class), any(AmazonS3Client.class));
    verifyNoMoreInteractionsHelper();
}
 

/** Unsupported Operation. */
@Override public SetObjectTaggingResult setObjectTagging(SetObjectTaggingRequest setObjTaggingReq) {
    throw new UnsupportedOperationException("Operation not supported");
}
 

/**
 * Set the tags for the specified object.
 *
 * @param setObjectTaggingRequest the request object containing all the options for setting the tags for the specified object
 * @param s3Client the {@link AmazonS3} implementation to use
 *
 * @return the result of the set operation
 */
public SetObjectTaggingResult setObjectTagging(SetObjectTaggingRequest setObjectTaggingRequest, AmazonS3 s3Client);