下面列出了怎么用com.amazonaws.services.s3.model.GetObjectTaggingResult的API类实例代码及写法，或者点击链接到github查看源代码。
@Test
public void testTagObjects()
{
// Create an S3 file transfer request parameters DTO to access S3 objects.
S3FileTransferRequestParamsDto params = new S3FileTransferRequestParamsDto();
params.setS3BucketName(S3_BUCKET_NAME);
// Create an S3 object summary.
S3ObjectSummary s3ObjectSummary = new S3ObjectSummary();
s3ObjectSummary.setKey(TARGET_S3_KEY);
// Create an S3 object tag.
Tag tag = new Tag(S3_OBJECT_TAG_KEY, S3_OBJECT_TAG_VALUE);
// Put a file in S3.
s3Operations.putObject(new PutObjectRequest(S3_BUCKET_NAME, TARGET_S3_KEY, new ByteArrayInputStream(new byte[1]), new ObjectMetadata()), null);
// Tag the file with an S3 object tag.
s3Dao.tagObjects(params, new S3FileTransferRequestParamsDto(), Collections.singletonList(s3ObjectSummary), tag);
// Validate that the object got tagged.
GetObjectTaggingResult getObjectTaggingResult = s3Operations.getObjectTagging(new GetObjectTaggingRequest(S3_BUCKET_NAME, TARGET_S3_KEY), null);
assertEquals(Collections.singletonList(tag), getObjectTaggingResult.getTagSet());
}
@Test
public void testTagObjectsOtherTagKeyAlreadyExists()
{
// Create two S3 object tags having different tag keys.
List<Tag> tags = Arrays.asList(new Tag(S3_OBJECT_TAG_KEY, S3_OBJECT_TAG_VALUE), new Tag(S3_OBJECT_TAG_KEY_2, S3_OBJECT_TAG_VALUE_2));
// Put a file in S3 that is already tagged with the first S3 object tag.
PutObjectRequest putObjectRequest = new PutObjectRequest(S3_BUCKET_NAME, TARGET_S3_KEY, new ByteArrayInputStream(new byte[1]), new ObjectMetadata());
putObjectRequest.setTagging(new ObjectTagging(Collections.singletonList(tags.get(0))));
s3Operations.putObject(putObjectRequest, null);
// Validate that the S3 object is tagged with the first tag only.
GetObjectTaggingResult getObjectTaggingResult = s3Operations.getObjectTagging(new GetObjectTaggingRequest(S3_BUCKET_NAME, TARGET_S3_KEY), null);
assertEquals(Collections.singletonList(tags.get(0)), getObjectTaggingResult.getTagSet());
// Create an S3 file transfer request parameters DTO to access S3 objects.
S3FileTransferRequestParamsDto params = new S3FileTransferRequestParamsDto();
params.setS3BucketName(S3_BUCKET_NAME);
// Create an S3 object summary.
S3ObjectSummary s3ObjectSummary = new S3ObjectSummary();
s3ObjectSummary.setKey(TARGET_S3_KEY);
// Tag the S3 file with the second S3 object tag.
s3Dao.tagObjects(params, new S3FileTransferRequestParamsDto(), Collections.singletonList(s3ObjectSummary), tags.get(1));
// Validate that the S3 object is now tagged with both tags.
getObjectTaggingResult = s3Operations.getObjectTagging(new GetObjectTaggingRequest(S3_BUCKET_NAME, TARGET_S3_KEY), null);
assertEquals(tags.size(), getObjectTaggingResult.getTagSet().size());
assertTrue(getObjectTaggingResult.getTagSet().containsAll(tags));
}
@Test
public void testTagObjectsTargetTagKeyAlreadyExists()
{
// Create two S3 object tags having the same tag key.
List<Tag> tags = Arrays.asList(new Tag(S3_OBJECT_TAG_KEY, S3_OBJECT_TAG_VALUE), new Tag(S3_OBJECT_TAG_KEY, S3_OBJECT_TAG_VALUE_2));
// Put a file in S3 that is already tagged with the first S3 object tag.
PutObjectRequest putObjectRequest = new PutObjectRequest(S3_BUCKET_NAME, TARGET_S3_KEY, new ByteArrayInputStream(new byte[1]), new ObjectMetadata());
putObjectRequest.setTagging(new ObjectTagging(Collections.singletonList(tags.get(0))));
s3Operations.putObject(putObjectRequest, null);
// Validate that the S3 object is tagged with the first tag.
GetObjectTaggingResult getObjectTaggingResult = s3Operations.getObjectTagging(new GetObjectTaggingRequest(S3_BUCKET_NAME, TARGET_S3_KEY), null);
assertEquals(Collections.singletonList(tags.get(0)), getObjectTaggingResult.getTagSet());
// Create an S3 file transfer request parameters DTO to access S3 objects.
S3FileTransferRequestParamsDto params = new S3FileTransferRequestParamsDto();
params.setS3BucketName(S3_BUCKET_NAME);
// Create an S3 object summary.
S3ObjectSummary s3ObjectSummary = new S3ObjectSummary();
s3ObjectSummary.setKey(TARGET_S3_KEY);
// Tag the S3 file with the second S3 object tag.
s3Dao.tagObjects(params, new S3FileTransferRequestParamsDto(), Collections.singletonList(s3ObjectSummary), tags.get(1));
// Validate that the S3 object is tagged with the second tag now.
getObjectTaggingResult = s3Operations.getObjectTagging(new GetObjectTaggingRequest(S3_BUCKET_NAME, TARGET_S3_KEY), null);
assertEquals(Collections.singletonList(tags.get(1)), getObjectTaggingResult.getTagSet());
}
@Test
public void testTagVersionsOtherTagKeyAlreadyExists()
{
// Create two S3 object tags having different tag keys.
List<Tag> tags = Arrays.asList(new Tag(S3_OBJECT_TAG_KEY, S3_OBJECT_TAG_VALUE), new Tag(S3_OBJECT_TAG_KEY_2, S3_OBJECT_TAG_VALUE_2));
// Put an S3 file that is already tagged with the first S3 object tag in an S3 bucket that has versioning enabled.
PutObjectRequest putObjectRequest =
new PutObjectRequest(MockS3OperationsImpl.MOCK_S3_BUCKET_NAME_VERSIONING_ENABLED, TARGET_S3_KEY, new ByteArrayInputStream(new byte[1]),
new ObjectMetadata());
putObjectRequest.setTagging(new ObjectTagging(Collections.singletonList(tags.get(0))));
s3Operations.putObject(putObjectRequest, null);
// List S3 versions that match the test S3 key.
ListVersionsRequest listVersionsRequest =
new ListVersionsRequest().withBucketName(MockS3OperationsImpl.MOCK_S3_BUCKET_NAME_VERSIONING_ENABLED).withPrefix(TARGET_S3_KEY);
VersionListing versionListing = s3Operations.listVersions(listVersionsRequest, null);
assertEquals(1, CollectionUtils.size(versionListing.getVersionSummaries()));
assertEquals(TARGET_S3_KEY, versionListing.getVersionSummaries().get(0).getKey());
assertNotNull(versionListing.getVersionSummaries().get(0).getVersionId());
// Validate that the S3 object is tagged with the first tag only.
GetObjectTaggingResult getObjectTaggingResult = s3Operations.getObjectTagging(
new GetObjectTaggingRequest(MockS3OperationsImpl.MOCK_S3_BUCKET_NAME_VERSIONING_ENABLED, TARGET_S3_KEY,
versionListing.getVersionSummaries().get(0).getVersionId()), null);
assertEquals(Collections.singletonList(tags.get(0)), getObjectTaggingResult.getTagSet());
// Tag the S3 version with the second S3 object tag.
S3FileTransferRequestParamsDto params = new S3FileTransferRequestParamsDto();
params.setS3BucketName(MockS3OperationsImpl.MOCK_S3_BUCKET_NAME_VERSIONING_ENABLED);
s3Dao.tagVersions(params, new S3FileTransferRequestParamsDto(), versionListing.getVersionSummaries(), tags.get(1));
// Validate that the S3 object is now tagged with both tags.
getObjectTaggingResult = s3Operations.getObjectTagging(
new GetObjectTaggingRequest(MockS3OperationsImpl.MOCK_S3_BUCKET_NAME_VERSIONING_ENABLED, TARGET_S3_KEY,
versionListing.getVersionSummaries().get(0).getVersionId()), null);
assertEquals(tags.size(), getObjectTaggingResult.getTagSet().size());
assertTrue(getObjectTaggingResult.getTagSet().containsAll(tags));
}
@Test
public void testTagVersionsS3BucketWithVersioningDisabled()
{
// Create an S3 object tag.
Tag tag = new Tag(S3_OBJECT_TAG_KEY, S3_OBJECT_TAG_VALUE);
// Put S3 objects in S3 bucket that has versioning disabled.
for (int i = 0; i < 2; i++)
{
s3Operations.putObject(new PutObjectRequest(S3_BUCKET_NAME, TARGET_S3_KEY + i, new ByteArrayInputStream(new byte[1]), new ObjectMetadata()), null);
}
// List S3 versions that match the test prefix.
ListVersionsRequest listVersionsRequest = new ListVersionsRequest().withBucketName(S3_BUCKET_NAME).withPrefix(TARGET_S3_KEY);
VersionListing versionListing = s3Operations.listVersions(listVersionsRequest, null);
assertEquals(2, CollectionUtils.size(versionListing.getVersionSummaries()));
for (int i = 0; i < 2; i++)
{
assertNull(versionListing.getVersionSummaries().get(i).getVersionId());
}
// Create an S3 file transfer request parameters DTO to access S3 objects.
S3FileTransferRequestParamsDto params = new S3FileTransferRequestParamsDto();
params.setS3BucketName(S3_BUCKET_NAME);
// Tag listed S3 versions with an S3 object tag.
s3Dao.tagVersions(params, new S3FileTransferRequestParamsDto(), versionListing.getVersionSummaries(), tag);
// Validate that both S3 objects got tagged.
for (int i = 0; i < 2; i++)
{
GetObjectTaggingResult getObjectTaggingResult =
s3Operations.getObjectTagging(new GetObjectTaggingRequest(S3_BUCKET_NAME, TARGET_S3_KEY + i, null), null);
assertEquals(Collections.singletonList(tag), getObjectTaggingResult.getTagSet());
}
}
@Test
public void testTagVersionsS3BucketWithVersioningEnabled()
{
// Create an S3 object tag.
Tag tag = new Tag(S3_OBJECT_TAG_KEY, S3_OBJECT_TAG_VALUE);
// Put two S3 versions in S3 bucket that has versioning enabled.
for (int i = 0; i < 2; i++)
{
s3Operations.putObject(
new PutObjectRequest(MockS3OperationsImpl.MOCK_S3_BUCKET_NAME_VERSIONING_ENABLED, TARGET_S3_KEY, new ByteArrayInputStream(new byte[1]),
new ObjectMetadata()), null);
}
// List S3 versions that match the test key.
ListVersionsRequest listVersionsRequest =
new ListVersionsRequest().withBucketName(MockS3OperationsImpl.MOCK_S3_BUCKET_NAME_VERSIONING_ENABLED).withPrefix(TARGET_S3_KEY);
VersionListing versionListing = s3Operations.listVersions(listVersionsRequest, null);
assertEquals(2, CollectionUtils.size(versionListing.getVersionSummaries()));
for (int i = 0; i < 2; i++)
{
assertNotNull(versionListing.getVersionSummaries().get(i).getVersionId());
}
// Create an S3 file transfer request parameters DTO to access S3 objects with a mocked S3 bucket name that enables S3 bucket versioning.
S3FileTransferRequestParamsDto params = new S3FileTransferRequestParamsDto();
params.setS3BucketName(MockS3OperationsImpl.MOCK_S3_BUCKET_NAME_VERSIONING_ENABLED);
// Tag listed S3 version with an S3 object tag.
s3Dao.tagVersions(params, new S3FileTransferRequestParamsDto(), versionListing.getVersionSummaries(), tag);
// Validate that both versions got tagged.
for (int i = 0; i < 2; i++)
{
GetObjectTaggingResult getObjectTaggingResult = s3Operations.getObjectTagging(
new GetObjectTaggingRequest(MockS3OperationsImpl.MOCK_S3_BUCKET_NAME_VERSIONING_ENABLED, TARGET_S3_KEY,
versionListing.getVersionSummaries().get(i).getVersionId()), null);
assertEquals(Collections.singletonList(tag), getObjectTaggingResult.getTagSet());
}
}
@Test
public void testTagVersionsTargetTagKeyAlreadyExists()
{
// Create two S3 object tags having the same tag key.
List<Tag> tags = Arrays.asList(new Tag(S3_OBJECT_TAG_KEY, S3_OBJECT_TAG_VALUE), new Tag(S3_OBJECT_TAG_KEY, S3_OBJECT_TAG_VALUE_2));
// Put an S3 file that is already tagged with the first S3 object tag in an S3 bucket that has versioning enabled.
PutObjectRequest putObjectRequest =
new PutObjectRequest(MockS3OperationsImpl.MOCK_S3_BUCKET_NAME_VERSIONING_ENABLED, TARGET_S3_KEY, new ByteArrayInputStream(new byte[1]),
new ObjectMetadata());
putObjectRequest.setTagging(new ObjectTagging(Collections.singletonList(tags.get(0))));
s3Operations.putObject(putObjectRequest, null);
// List S3 versions that match the test S3 key.
ListVersionsRequest listVersionsRequest =
new ListVersionsRequest().withBucketName(MockS3OperationsImpl.MOCK_S3_BUCKET_NAME_VERSIONING_ENABLED).withPrefix(TARGET_S3_KEY);
VersionListing versionListing = s3Operations.listVersions(listVersionsRequest, null);
assertEquals(1, CollectionUtils.size(versionListing.getVersionSummaries()));
assertEquals(TARGET_S3_KEY, versionListing.getVersionSummaries().get(0).getKey());
assertNotNull(versionListing.getVersionSummaries().get(0).getVersionId());
// Validate that the S3 object is tagged with the first tag only.
GetObjectTaggingResult getObjectTaggingResult = s3Operations.getObjectTagging(
new GetObjectTaggingRequest(MockS3OperationsImpl.MOCK_S3_BUCKET_NAME_VERSIONING_ENABLED, TARGET_S3_KEY,
versionListing.getVersionSummaries().get(0).getVersionId()), null);
assertEquals(Collections.singletonList(tags.get(0)), getObjectTaggingResult.getTagSet());
// Create an S3 file transfer request parameters DTO to access S3 objects.
S3FileTransferRequestParamsDto params = new S3FileTransferRequestParamsDto();
params.setS3BucketName(MockS3OperationsImpl.MOCK_S3_BUCKET_NAME_VERSIONING_ENABLED);
// Tag the S3 version with the second S3 object tag.
s3Dao.tagVersions(params, new S3FileTransferRequestParamsDto(), versionListing.getVersionSummaries(), tags.get(1));
// Validate that the S3 object is now tagged with the second tag only.
getObjectTaggingResult = s3Operations.getObjectTagging(
new GetObjectTaggingRequest(MockS3OperationsImpl.MOCK_S3_BUCKET_NAME_VERSIONING_ENABLED, TARGET_S3_KEY,
versionListing.getVersionSummaries().get(0).getVersionId()), null);
assertEquals(Collections.singletonList(tags.get(1)), getObjectTaggingResult.getTagSet());
}
private Record createRecordForListing(final S3VersionSummary versionSummary, final GetObjectTaggingResult taggingResult, final ObjectMetadata objectMetadata) {
final Map<String, Object> values = new HashMap<>();
values.put(KEY, versionSummary.getKey());
values.put(BUCKET, versionSummary.getBucketName());
if (versionSummary.getOwner() != null) { // We may not have permission to read the owner
values.put(OWNER, versionSummary.getOwner().getId());
}
values.put(ETAG, versionSummary.getETag());
values.put(LAST_MODIFIED, new Timestamp(versionSummary.getLastModified().getTime()));
values.put(SIZE, versionSummary.getSize());
values.put(STORAGE_CLASS, versionSummary.getStorageClass());
values.put(IS_LATEST, versionSummary.isLatest());
final String versionId = versionSummary.getVersionId();
if (versionId != null && !versionId.equals(Constants.NULL_VERSION_ID)) {
values.put(VERSION_ID, versionSummary.getVersionId());
}
if (taggingResult != null) {
final Map<String, String> tags = new HashMap<>();
taggingResult.getTagSet().forEach(tag -> {
tags.put(tag.getKey(), tag.getValue());
});
values.put(TAGS, tags);
}
if (objectMetadata != null) {
values.put(USER_METADATA, objectMetadata.getUserMetadata());
}
return new MapRecord(RECORD_SCHEMA, values);
}
@Override
public void addToListing(final S3VersionSummary versionSummary, final GetObjectTaggingResult taggingResult, final ObjectMetadata objectMetadata) {
// Create the attributes
final Map<String, String> attributes = new HashMap<>();
attributes.put(CoreAttributes.FILENAME.key(), versionSummary.getKey());
attributes.put("s3.bucket", versionSummary.getBucketName());
if (versionSummary.getOwner() != null) { // We may not have permission to read the owner
attributes.put("s3.owner", versionSummary.getOwner().getId());
}
attributes.put("s3.etag", versionSummary.getETag());
attributes.put("s3.lastModified", String.valueOf(versionSummary.getLastModified().getTime()));
attributes.put("s3.length", String.valueOf(versionSummary.getSize()));
attributes.put("s3.storeClass", versionSummary.getStorageClass());
attributes.put("s3.isLatest", String.valueOf(versionSummary.isLatest()));
if (versionSummary.getVersionId() != null) {
attributes.put("s3.version", versionSummary.getVersionId());
}
if (taggingResult != null) {
final List<Tag> tags = taggingResult.getTagSet();
for (final Tag tag : tags) {
attributes.put("s3.tag." + tag.getKey(), tag.getValue());
}
}
if (objectMetadata != null) {
for (Map.Entry<String, String> e : objectMetadata.getUserMetadata().entrySet()) {
attributes.put("s3.user.metadata." + e.getKey(), e.getValue());
}
}
// Create the flowfile
FlowFile flowFile = session.create();
flowFile = session.putAllAttributes(flowFile, attributes);
session.transfer(flowFile, REL_SUCCESS);
}
@Test
public void testObjectTags() throws IOException, InterruptedException {
final TestRunner runner = TestRunners.newTestRunner(new PutS3Object());
runner.setProperty(PutS3Object.CREDENTIALS_FILE, CREDENTIALS_FILE);
runner.setProperty(PutS3Object.REGION, REGION);
runner.setProperty(PutS3Object.BUCKET, BUCKET_NAME);
runner.setProperty(PutS3Object.OBJECT_TAGS_PREFIX, "tagS3");
runner.setProperty(PutS3Object.REMOVE_TAG_PREFIX, "true");
final Map<String, String> attrs = new HashMap<>();
attrs.put("filename", "tag-test.txt");
attrs.put("tagS3PII", "true");
runner.enqueue(getResourcePath(SAMPLE_FILE_RESOURCE_NAME), attrs);
runner.run();
runner.assertAllFlowFilesTransferred(PutS3Object.REL_SUCCESS, 1);
GetObjectTaggingResult result = client.getObjectTagging(new GetObjectTaggingRequest(BUCKET_NAME, "tag-test.txt"));
List<Tag> objectTags = result.getTagSet();
for (Tag tag : objectTags) {
System.out.println("Tag Key : " + tag.getKey() + ", Tag Value : " + tag.getValue());
}
Assert.assertTrue(objectTags.size() == 1);
Assert.assertEquals("PII", objectTags.get(0).getKey());
Assert.assertEquals("true", objectTags.get(0).getValue());
}
@Test
public void testSimpleTag() throws Exception {
String objectKey = "test-file";
String tagKey = "nifi-key";
String tagValue = "nifi-val";
// put file in s3
putTestFile(objectKey, getFileFromResourceName(SAMPLE_FILE_RESOURCE_NAME));
// Set up processor
final TestRunner runner = TestRunners.newTestRunner(new TagS3Object());
runner.setProperty(TagS3Object.CREDENTIALS_FILE, CREDENTIALS_FILE);
runner.setProperty(TagS3Object.REGION, REGION);
runner.setProperty(TagS3Object.BUCKET, BUCKET_NAME);
runner.setProperty(TagS3Object.TAG_KEY, tagKey);
runner.setProperty(TagS3Object.TAG_VALUE, tagValue);
final Map<String, String> attrs = new HashMap<>();
attrs.put("filename", objectKey);
runner.enqueue(new byte[0], attrs);
// tag file
runner.run(1);
// Verify processor succeeds
runner.assertAllFlowFilesTransferred(TagS3Object.REL_SUCCESS, 1);
// Verify tag exists on S3 object
GetObjectTaggingResult res = client.getObjectTagging(new GetObjectTaggingRequest(BUCKET_NAME, objectKey));
assertTrue("Expected tag not found on S3 object", res.getTagSet().contains(new Tag(tagKey, tagValue)));
}
@Test
public void testAppendTag() throws Exception {
String objectKey = "test-file";
String tagKey = "nifi-key";
String tagValue = "nifi-val";
Tag existingTag = new Tag("oldkey", "oldvalue");
// put file in s3
putFileWithObjectTag(objectKey, getFileFromResourceName(SAMPLE_FILE_RESOURCE_NAME), Arrays.asList(existingTag));
// Set up processor
final TestRunner runner = TestRunners.newTestRunner(new TagS3Object());
runner.setProperty(TagS3Object.CREDENTIALS_FILE, CREDENTIALS_FILE);
runner.setProperty(TagS3Object.REGION, REGION);
runner.setProperty(TagS3Object.BUCKET, BUCKET_NAME);
runner.setProperty(TagS3Object.TAG_KEY, tagKey);
runner.setProperty(TagS3Object.TAG_VALUE, tagValue);
final Map<String, String> attrs = new HashMap<>();
attrs.put("filename", objectKey);
runner.enqueue(new byte[0], attrs);
// tag file
runner.run(1);
// Verify processor succeeds
runner.assertAllFlowFilesTransferred(TagS3Object.REL_SUCCESS, 1);
// Verify new tag and existing exist on S3 object
GetObjectTaggingResult res = client.getObjectTagging(new GetObjectTaggingRequest(BUCKET_NAME, objectKey));
assertTrue("Expected new tag not found on S3 object", res.getTagSet().contains(new Tag(tagKey, tagValue)));
assertTrue("Expected existing tag not found on S3 object", res.getTagSet().contains(existingTag));
}
public void processDirectory(AmazonS3 s3, ObjectListing listing, String bucket) throws Exception {
double time = System.currentTimeMillis();
ExecutorService executor = Executors.newFixedThreadPool(16);
int count = 0;
for (S3ObjectSummary objectSummary : listing.getObjectSummaries()) {
if ("STANDARD".equalsIgnoreCase(objectSummary.getStorageClass())) {
long size = objectSummary.getSize();
logger.debug("*** Processing S3 {}, size: {}", objectSummary.getKey(), size);
S3Object object = s3.getObject(new GetObjectRequest(bucket, objectSummary.getKey()));
String bucketName = object.getBucketName();
String keyName = object.getKey();
GetObjectTaggingRequest request = new GetObjectTaggingRequest(bucketName, keyName);
GetObjectTaggingResult result = s3.getObjectTagging(request);
List<Tag> tags = result.getTagSet();
String type = null;
String name = null;
if (tags.isEmpty()) {
object.close();
logger.warn("Error, S3 object: {} has no tags", keyName);
} else {
for (Tag tag : tags) {
String key = tag.getKey();
String value = tag.getValue();
if (key.equals("type")) {
type = value;
}
if (key.equals("name")) {
name = value;
}
}
if (name == null) {
object.close();
throw new Exception("Error: " + keyName + " is missing a name tag");
}
if (name.contains(" ")) {
object.close();
throw new Exception("Error: " + keyName + " has a name attribute with a space in it");
}
if (type == null) {
object.close();
throw new Exception("Error: " + keyName + " has no type tag");
}
if (!name.startsWith("$"))
name = "$" + name;
// The runnable will call object.close();
Runnable w = new AwsWorker(type, name, object, size);
executor.execute(w);
count++;
}
}
}
executor.shutdown();
executor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
time = System.currentTimeMillis() - time;
time = time / 60000;
logger.info("Initialized all {} S3 objects in {} minutes", count, time);
}
@Override
public GetObjectTaggingResult getObjectTagging(GetObjectTaggingRequest getObjectTaggingRequest, AmazonS3 s3Client)
{
return s3Client.getObjectTagging(getObjectTaggingRequest);
}
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();
}
}
}
@Override
public GetObjectTaggingResult getObjectTagging(GetObjectTaggingRequest getObjectTaggingRequest, AmazonS3 s3Client)
{
return new GetObjectTaggingResult(
getMockS3Object(getObjectTaggingRequest.getBucketName(), getObjectTaggingRequest.getKey(), getObjectTaggingRequest.getVersionId()).getTags());
}
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 GetObjectTaggingResult getObjectTagging(GetObjectTaggingRequest getObjTaggingReq) {
throw new UnsupportedOperationException("Operation not supported");
}
public void processDirectory(AmazonS3Client s3, ObjectListing listing, String bucket) throws Exception {
for (S3ObjectSummary objectSummary : listing.getObjectSummaries()) {
long size = objectSummary.getSize();
logger.info("*** Processing S3 {}, size: {}",objectSummary.getKey(),size);
S3Object object = s3.getObject(new GetObjectRequest(bucket, objectSummary.getKey()));
String bucketName = object.getBucketName();
String keyName = object.getKey();
GetObjectTaggingRequest request = new GetObjectTaggingRequest(bucketName, keyName);
GetObjectTaggingResult result = s3.getObjectTagging(request);
List<Tag> tags = result.getTagSet();
String type = null;
String name = null;
if (tags.isEmpty()) {
System.err.println("Error: " + keyName + " has no tags");
} else {
for (Tag tag : tags) {
String key = tag.getKey();
String value = tag.getValue();
if (key.equals("type")) {
type = value;
}
if (key.equals("name")) {
name = value;
}
}
if (name == null)
throw new Exception("Error: " + keyName + " is missing a name tag");
if (name.contains(" "))
throw new Exception("Error: " + keyName + " has a name attribute with a space in it");
if (type == null)
throw new Exception("Error: " + keyName + " has no type tag");
if (!name.startsWith("$"))
name = "$" + name;
readData(type, name, object, size);
}
}
}
@Override
public void addToListing(final S3VersionSummary summary, final GetObjectTaggingResult taggingResult, final ObjectMetadata objectMetadata) throws IOException {
recordWriter.write(createRecordForListing(summary, taggingResult, objectMetadata));
}
@Test
public void testReplaceTags() throws Exception {
String objectKey = "test-file";
String tagKey = "nifi-key";
String tagValue = "nifi-val";
Tag existingTag = new Tag("s3.tag.oldkey", "oldvalue");
// put file in s3
putFileWithObjectTag(objectKey, getFileFromResourceName(SAMPLE_FILE_RESOURCE_NAME), Arrays.asList(existingTag));
// Set up processor
final TestRunner runner = TestRunners.newTestRunner(new TagS3Object());
runner.setProperty(TagS3Object.CREDENTIALS_FILE, CREDENTIALS_FILE);
runner.setProperty(TagS3Object.REGION, REGION);
runner.setProperty(TagS3Object.BUCKET, BUCKET_NAME);
runner.setProperty(TagS3Object.TAG_KEY, tagKey);
runner.setProperty(TagS3Object.TAG_VALUE, tagValue);
runner.setProperty(TagS3Object.APPEND_TAG, "false");
final Map<String, String> attrs = new HashMap<>();
attrs.put("filename", objectKey);
attrs.put("s3.tag."+existingTag.getKey(), existingTag.getValue());
runner.enqueue(new byte[0], attrs);
// tag file
runner.run(1);
// Verify processor succeeds
runner.assertAllFlowFilesTransferred(TagS3Object.REL_SUCCESS, 1);
// Verify flowfile attributes match s3 tags
MockFlowFile flowFiles = runner.getFlowFilesForRelationship(TagS3Object.REL_SUCCESS).get(0);
flowFiles.assertAttributeNotExists(existingTag.getKey());
flowFiles.assertAttributeEquals("s3.tag."+tagKey, tagValue);
// Verify new tag exists on S3 object and prior tag removed
GetObjectTaggingResult res = client.getObjectTagging(new GetObjectTaggingRequest(BUCKET_NAME, objectKey));
assertTrue("Expected new tag not found on S3 object", res.getTagSet().contains(new Tag(tagKey, tagValue)));
assertFalse("Existing tag not replaced on S3 object", res.getTagSet().contains(existingTag));
}
private void mockGetExistingTags(Tag... currentTag) {
List<Tag> currentTags = new ArrayList<>(Arrays.asList(currentTag));
Mockito.when(mockS3Client.getObjectTagging(Mockito.any())).thenReturn(new GetObjectTaggingResult(currentTags));
}
public static void main(String[] args) {
if (args.length < 2) {
System.out.println("Please specify a bucket name and key name");
System.exit(1);
}
// snippet-start:[s3.java.getobjecttags.main]
String bucketName = args[0];
String keyName = args[1];
System.out.println("Retrieving Object Tags for " + keyName);
final AmazonS3 s3 = AmazonS3ClientBuilder.standard().withRegion(Regions.DEFAULT_REGION).build();
try {
GetObjectTaggingRequest getTaggingRequest = new GetObjectTaggingRequest(bucketName, keyName);
GetObjectTaggingResult tags = s3.getObjectTagging(getTaggingRequest);
List<Tag> tagSet= tags.getTagSet();
//Iterate through the list
Iterator<Tag> tagIterator = tagSet.iterator();
while(tagIterator.hasNext()) {
Tag tag = (Tag)tagIterator.next();
System.out.println(tag.getKey());
System.out.println(tag.getValue());
}
} catch (AmazonServiceException e) {
System.err.println(e.getErrorMessage());
System.exit(1);
}
// snippet-end:[s3.java.getobjecttags.main]
}
/**
* Returns all S3 object tags for the specified object.
*
* @param getObjectTaggingRequest the request object containing all the options on how to retrieve the Amazon S3 object tags
* @param s3Client the {@link AmazonS3} implementation to use
*
* @return the set of S3 object tags
*/
public GetObjectTaggingResult getObjectTagging(GetObjectTaggingRequest getObjectTaggingRequest, AmazonS3 s3Client);
void addToListing(S3VersionSummary summary, GetObjectTaggingResult taggingResult, ObjectMetadata objectMetadata) throws IOException;