Skip to content
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

Bug Fix: S3 source key #1926

Merged
merged 2 commits into from
Oct 19, 2022
Merged

Bug Fix: S3 source key #1926

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
740e709
Fix: S3 source key bug fix
asifsmohammed Oct 15, 2022
a8957d8
Fix: Added unit test
asifsmohammed Oct 15, 2022
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
4 changes: 2 additions & 2 deletions ...ource/src/integrationTest/java/org/opensearch/dataprepper/plugins/source/SqsWorkerIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,7 +107,7 @@ void processSqsMessages_should_return_at_least_one_message(final int numberOfObj
verify(s3Service, atLeastOnce()).addS3Object(s3ObjectReferenceArgumentCaptor.capture());

assertThat(s3ObjectReferenceArgumentCaptor.getValue().getBucketName(), equalTo(bucket));
assertThat(s3ObjectReferenceArgumentCaptor.getValue().getKey(), startsWith("s3source/sqs/"));
assertThat(s3ObjectReferenceArgumentCaptor.getValue().getKey(), startsWith("s3 source/sqs/"));
assertThat(sqsMessagesProcessed, greaterThanOrEqualTo(1));
assertThat(sqsMessagesProcessed, lessThanOrEqualTo(numberOfObjectsToWrite));
}
Expand All @@ -124,7 +124,7 @@ private void writeToS3(final int numberOfObjectsToWrite) throws IOException {
final int numberOfRecords = 100;
final NewlineDelimitedRecordsGenerator newlineDelimitedRecordsGenerator = new NewlineDelimitedRecordsGenerator();
for (int i = 0; i < numberOfObjectsToWrite; i++) {
final String key = "s3source/sqs/" + UUID.randomUUID() + "_" + Instant.now().toString() + newlineDelimitedRecordsGenerator.getFileExtension();
final String key = "s3 source/sqs/" + UUID.randomUUID() + "_" + Instant.now().toString() + newlineDelimitedRecordsGenerator.getFileExtension();
// isCompressionEnabled is set to false since we test for compression in S3ObjectWorkerIT
s3ObjectGenerator.write(numberOfRecords, key, newlineDelimitedRecordsGenerator, false);
}
Expand Down
2 changes: 1 addition & 1 deletion ...-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/SqsWorker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -219,7 +219,7 @@ private boolean isEventNameCreated(final S3EventNotification.S3EventNotification
private S3ObjectReference populateS3Reference(final S3EventNotification.S3EventNotificationRecord s3EventNotificationRecord) {
final S3EventNotification.S3Entity s3Entity = s3EventNotificationRecord.getS3();
return S3ObjectReference.bucketAndKey(s3Entity.getBucket().getName(),
s3Entity.getObject().getKey())
s3Entity.getObject().getUrlDecodedKey())
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a static method in the model class. And since #1628 it has been in our code. I think it might be better to continue to use getKey, and move the code for getUrlDecodedKey() into S3Worker. That will make it easier to test.

I'm fine with following on with this later as well.

.build();
}
}
28 changes: 28 additions & 0 deletions ...gins/s3-source/src/test/java/org/opensearch/dataprepper/plugins/source/SqsWorkerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,8 @@
import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse;
import software.amazon.awssdk.services.sqs.model.SqsException;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.time.Duration;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
Expand Down Expand Up @@ -245,4 +247,30 @@ void processSqsMessages_should_invoke_delete_if_input_is_not_valid_JSON_and_dele
verify(sqsMessagesDeletedCounter).increment(1);
verify(sqsMessagesFailedCounter).increment();
}

@Test
void populateS3Reference_should_interact_with_getUrlDecodedKey() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we not test an s3Entity with a key that has a space in it? Why do we need reflection?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I tried testing it, I didn't find a way to create an instance of S3EventNotificationRecord without using mocks. So with the mocks we are just returning whatever we define in when then condition.

I used refection here to test a private method. I'm sure there are ways to improve the logic here.

// Using reflection to unit test a private method as part of bug fix.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minor and non-blocking: It makes sense to include a link to the Bug for future reference

final Method method = SqsWorker.class.getDeclaredMethod("populateS3Reference", S3EventNotification.S3EventNotificationRecord.class);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You can also make this method package protected specifically for testing purposes. I do think there is probably some opportunity for refactoring, but this should work for now.

method.setAccessible(true);

final S3EventNotification.S3EventNotificationRecord s3EventNotificationRecord = mock(S3EventNotification.S3EventNotificationRecord.class);
final S3EventNotification.S3Entity s3Entity = mock(S3EventNotification.S3Entity.class);
final S3EventNotification.S3ObjectEntity s3ObjectEntity = mock(S3EventNotification.S3ObjectEntity.class);
final S3EventNotification.S3BucketEntity s3BucketEntity = mock(S3EventNotification.S3BucketEntity.class);

when(s3EventNotificationRecord.getS3()).thenReturn(s3Entity);
when(s3Entity.getBucket()).thenReturn(s3BucketEntity);
when(s3Entity.getObject()).thenReturn(s3ObjectEntity);
when(s3BucketEntity.getName()).thenReturn("test-bucket-name");
when(s3ObjectEntity.getUrlDecodedKey()).thenReturn("test-key");

final S3ObjectReference s3ObjectReference = (S3ObjectReference) method.invoke(sqsWorker, s3EventNotificationRecord);

assertThat(s3ObjectReference, notNullValue());
assertThat(s3ObjectReference.getBucketName(), equalTo("test-bucket-name"));
assertThat(s3ObjectReference.getKey(), equalTo("test-key"));
verify(s3ObjectEntity).getUrlDecodedKey();
verifyNoMoreInteractions(s3ObjectEntity);
}
}