Adding testing architecture and KCL 2.x basic polling/streaming tests #1136

mmankika-aws · 2023-06-09T18:38:55Z

Issue #, if available:

Description of changes:
Adding testing architecture to implement automated integration tests.
Added KCL2XIntegrationTests which implements basic polling and streaming consumer tests against a single producer.
Integration tests require AWS credentials to be provided. The integration tests can be run with the command
mvn -Dit.test=KCLV2IntegrationTest -Dcredentials="<IAM_USER>" verify

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

…ests

stair-aws

Partial review; need to divert to other work. To expedite the review process, please apply any learnings/comments to all code (which may not have been reviewed yet).

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/RecordValidatorQueue.java

...n-kinesis-client/src/test/java/software/amazon/kinesis/integration_tests/TestConsumerV2.java

...on-kinesis-client/src/test/java/software/amazon/kinesis/common/RecordValidatorQueueTest.java

...sis-client/src/test/java/software/amazon/kinesis/integration_tests/KCL2XIntegrationTest.java

...s-client/src/test/java/software/amazon/kinesis/integration_tests/StreamExistenceManager.java

stair-aws · 2023-06-13T15:09:01Z

...s-client/src/test/java/software/amazon/kinesis/integration_tests/StreamExistenceManager.java

+
+    public StreamExistenceManager(KCLAppConfig config) throws URISyntaxException, IOException {
+        this.testConfig = config;
+        this.client = config.buildAsyncKinesisClient(Protocol.HTTP1_1);


This ignores any protocol specified by the KCLAppConfig

This client is for the DescribeStreams API call. For this, it does not matter which HTTP protocol is being used so a default is put here. When making PutRecord, GetRecord calls, the Consumer Protocol specified in KCLAppConfig is used. However, I will update this so that they are consistent.

...s-client/src/test/java/software/amazon/kinesis/integration_tests/StreamExistenceManager.java

...on-kinesis-client/src/test/java/software/amazon/kinesis/common/RecordValidatorQueueTest.java

amazon-kinesis-client/pom.xml

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/RecordValidatorQueue.java

amazon-kinesis-client/src/test/java/software/amazon/kinesis/config/KCLAppConfig.java

...ient/src/test/java/software/amazon/kinesis/config/KCLReleaseCanary2XPollingH1TestConfig.java

ZeyuLi-AWS · 2023-06-13T07:07:59Z

...ient/src/test/java/software/amazon/kinesis/config/KCLReleaseCanary2XStreamingTestConfig.java

+        Instant instant = d.atZone( ZoneId.systemDefault() ).toInstant();
+        Date startStreamTime = Date.from( instant );
+
+        InitialPositionInStreamExtended initialPosition = InitialPositionInStreamExtended


FMI why we chose to start at (current time -5 min)?

We want to start processing records that were put onto the stream in this test case. This is just a small value that estimates approximately when the test case may have started (related to the time that the consumer has finished getting started and is now processing records).

amazon-kinesis-client/src/test/java/software/amazon/kinesis/config/KCLAppConfig.java

stair-aws

Partial review.

README.md

...on-kinesis-client/src/test/java/software/amazon/kinesis/config/BasicReleaseCanaryConfig.java

amazon-kinesis-client/src/test/java/software/amazon/kinesis/config/KCLAppConfig.java

...is-client/src/test/java/software/amazon/kinesis/config/ReleaseCanaryPollingH1TestConfig.java

...nt/src/test/java/software/amazon/kinesis/lifecycle/BasicStreamingPollingIntegrationTest.java

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/RecordValidatorQueueTest.java

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/TestConsumer.java

...is-client/src/test/java/software/amazon/kinesis/config/ReleaseCanaryPollingH1TestConfig.java

README.md

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/TestConsumer.java

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/RecordValidatorQueue.java

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/StreamExistenceManager.java

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/TestRecordProcessor.java

abhit17 · 2023-06-20T00:42:32Z

...nt/src/test/java/software/amazon/kinesis/lifecycle/BasicStreamingPollingIntegrationTest.java

+public class BasicStreamingPollingIntegrationTest {
+
+    @Test
+    public void KCLReleaseCanaryPollingH2Test() throws Exception {


method name should always start with small letters, also better naming convention for tests is "MethodName_StateUnderTest_ExpectedBehavior" ref https://medium.com/@stefanovskyi/unit-test-naming-conventions-dd9208eadbea

also better naming convention for tests is "MethodName_StateUnderTest_ExpectedBehavior" ref https://medium.com/@stefanovskyi/unit-test-naming-conventions-dd9208eadbea

The above statement is subjective.

Fact: Java method naming convention is lowerCamelCase.

From Javadoc wiki:

Writing comments and Javadoc is for better understanding the code and thus better maintaining it.

Opinion 1: Natural language (e.g., English) is much better at communicating intent than anAbridgedCamelCaseMashup, regardless of whether underscores are used as delimiters.
Opinion 2: The proposed method naming suggestion feels like a manifestation of Law of the instrument. (e.g., "If the only tool you have is a hammer, it is tempting to treat everything as if it were a nail")

Suggestion: if the test and intent cannot be made evident w/ a conventional method name, add Javadoc. Document the intent, preconditions, postconditions, and anything else that a future maintainer might find useful or informative. Natural language does not have compile-time constraints, and provides a much broader palette on which the author may convey details.

abhit17 · 2023-06-20T00:43:50Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/RecordValidatorQueueTest.java

+
+public class RecordValidatorQueueTest {
+
+    private RecordValidatorQueue recordValidator = new RecordValidatorQueue();


abhit17 · 2023-06-20T00:50:19Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/TestConsumer.java

+        // Setup configuration of KCL (including DynamoDB and CloudWatch)
+        DynamoDbAsyncClient dynamoClient = DynamoDbAsyncClient.builder().region(region).build();
+        CloudWatchAsyncClient cloudWatchClient = CloudWatchAsyncClient.builder().region(region).build();
+        ConfigsBuilder configsBuilder = new ConfigsBuilder(streamName, streamName, kinesisClient, dynamoClient, cloudWatchClient, UUID.randomUUID().toString(), new TestRecordProcessorFactory(recordValidator));


ConfigBuild creation logic is in KCLAppConfig already, re use that and avoid re-creating the same clients again. The clients like ddb kcl are created with credentials already, no need to create them again.

abhit17 · 2023-06-20T00:52:30Z

...n-kinesis-client/src/test/java/software/amazon/kinesis/utils/TestRecordProcessorFactory.java

+
+    RecordValidatorQueue recordValidator;
+
+    public TestRecordProcessorFactory(RecordValidatorQueue recordValidator) {


This gives a compile error

abhit17 · 2023-06-20T20:32:07Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/config/KCLAppConfig.java

+    public final KinesisAsyncClient buildAsyncKinesisClient(Protocol protocol) throws URISyntaxException, IOException {
+        if (kinesisAsyncClient == null) {
+            this.kinesisAsyncClient = buildAsyncKinesisClient(Optional.ofNullable(protocol));
+        }
+        return this.kinesisAsyncClient;


This is duplicate method, remove this.

abhit17 · 2023-06-20T20:53:22Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/TestRecordProcessor.java

+
+    private String shardId;
+
+    RecordValidatorQueue recordValidator;


private final

abhit17 · 2023-06-20T20:53:56Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/TestRecordProcessor.java

+        try {
+            log.info("Processing {} record(s)", processRecordsInput.records().size());
+
+            for (KinesisClientRecord r : processRecordsInput.records()) {


np r -> kinesisRecord or similar. Dont use single alphabet words

abhit17 · 2023-06-20T20:54:02Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/TestRecordProcessor.java

+            log.info("Processing {} record(s)", processRecordsInput.records().size());
+
+            for (KinesisClientRecord r : processRecordsInput.records()) {
+                String data = new String(asByteArray(r.data()));


amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/TestConsumer.java

abhit17 · 2023-06-20T20:58:38Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/RecordValidatorQueue.java

+                int nextVal = Integer.parseInt(record);
+                if (prevVal > nextVal) {
+                    log.error("The records are not in increasing order. Saw record data {} before {}.", prevVal, nextVal);
+                    shardIncOrder = false;


after this point, no need to check the next records of this shard or even other shards, test is going to fail anyways, Fail here directly and go back

abhit17 · 2023-06-21T16:46:52Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/RecordValidatorQueue.java

+    public RecordValidationStatus validateRecords(int expectedShardCount) {
+
+        // Validate that each List in the HashMap has data records in increasing order
+        boolean incOrder = true;


not used anymore, remove

abhit17 · 2023-06-21T16:47:30Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/RecordValidatorQueue.java

+        }
+
+        // If this is true, then there was some record that was missed during processing.
+        if (actualShardCount != expectedShardCount) {


this is not shardCount, these are record counts. Rename the variables

abhit17 · 2023-06-21T16:49:54Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/TestConsumer.java

+            validateRecordProcessor();
+
+            // Clean up resources created
+            Thread.sleep(TimeUnit.SECONDS.toMillis(30));


what is this delay added for?

There is sometimes a delay in the workers getting shutdown and deleted. This delay ensures that that process is complete before we try to delete any resources.

abhit17 · 2023-06-21T16:50:25Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/TestConsumer.java

+
+            // Clean up resources created
+            Thread.sleep(TimeUnit.SECONDS.toMillis(30));
+            deleteResources(streamExistenceManager, leaseTableManager);


move deleteResources to finally block, this is something you need to do in both try and catch block right

abhit17 · 2023-06-21T16:52:41Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/TestConsumer.java

+        } catch (InterruptedException e) {
+            log.info("Interrupted, assuming shutdown.");
+        } catch (ExecutionException | RuntimeException e) {
+            log.error("Error during publish records");


log exception as well

abhit17 · 2023-06-21T16:53:58Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/TestConsumer.java

+            log.error("Error creating payload data for {}", payloadCounter.toString());
+            throw new RuntimeException("Error converting object to bytes: ", e);


Skip logging of the error, you are throwing it so this should be logged once at top level

abhit17 · 2023-06-21T16:55:39Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/TestConsumer.java

+        } catch (InterruptedException e) {
+            log.info("Interrupted while waiting for graceful shutdown. Continuing.");
+        } catch (ExecutionException | TimeoutException e) {
+            throw new Exception("Exception while executing graceful shutdown. {}", e);


dont throw Exception of class Exception. The captured exception already has enough info just throw that same

abhit17 · 2023-06-21T16:58:04Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/LeaseTableManager.java

+        final ListTablesRequest request = ListTablesRequest.builder().build();
+        ListTablesResponse response = null;
+        while(response == null || response.lastEvaluatedTableName() != null) {
+            try {
+                response = FutureUtils.resolveOrCancelFuture(dynamoClient.listTables(request), Duration.ofSeconds(60));
+            } catch (ExecutionException | InterruptedException e) {


when you make a paginated call again, the request has to include the token from last call to make progress. Else this will keep on just getting same results with same request.

Also try to abstract the common code for this method as well and just keep API request creation and API call here.

Fixing the request that is made. With this new change there are many places in this method that are resource dependent (the initialization at the top, while condition, try block, addAll call...) that would need to be implemented in the resource specific managers that I think that abstracting this would harm readability more than it would help.

abhit17 · 2023-06-21T16:58:54Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/AWSResourceManager.java

+@Slf4j
+public abstract class AWSResourceManager {
+
+    public AWSResourceManager() {}


use lombok for this.

abhit17 · 2023-06-21T17:00:08Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/LeaseTableManager.java

+    public LeaseTableManager(DynamoDbAsyncClient dynamoClient) throws URISyntaxException, IOException {
+        this.dynamoClient = dynamoClient;


use lombok's @AllArgConstructor

Reviewed by abhit17@

Adding testing architecture and KCL 2.x basic polling and streaming t…

f7d286a

…ests

mmankika-aws requested review from abhit17, stair-aws, ZeyuLi-AWS and akidambisrinivasan June 9, 2023 18:38

mmankika-aws self-assigned this Jun 9, 2023

Removing internal references

d191401

stair-aws suggested changes Jun 13, 2023

View reviewed changes

stair-aws added the v2.x Issues related to the 2.x version label Jun 13, 2023

ZeyuLi-AWS reviewed Jun 13, 2023

View reviewed changes

Updates based on initial comments

37f6c04

mmankika-aws requested review from stair-aws and ZeyuLi-AWS June 13, 2023 22:25

mmankika-aws added 3 commits June 14, 2023 14:18

Bug fixes to testing architecture

67ebab7

Fixed producer cancelling before final task was finished

eb07782

Dead code removed

8dcbb75

mmankika-aws commented Jun 16, 2023

View reviewed changes

amazon-kinesis-client/src/test/java/software/amazon/kinesis/config/KCLAppConfig.java Outdated Show resolved Hide resolved

stair-aws previously requested changes Jun 16, 2023

View reviewed changes

mmankika-aws added 2 commits June 16, 2023 16:16

Updates based on 2nd round of comments

74a74c3

Fixing comment

ae42e96

mmankika-aws commented Jun 16, 2023

View reviewed changes

...is-client/src/test/java/software/amazon/kinesis/config/ReleaseCanaryPollingH1TestConfig.java Show resolved Hide resolved

mmankika-aws requested a review from stair-aws June 16, 2023 23:36

ZeyuLi-AWS reviewed Jun 19, 2023

View reviewed changes

mmankika-aws added 4 commits June 19, 2023 09:37

Few minor revisions

97d7831

Bug fix from last revision

ad26ac5

Revisions from review

23edc1b

Bug fix on lease table manager

2cfb411

abhit17 reviewed Jun 20, 2023

View reviewed changes

mmankika-aws added 2 commits June 20, 2023 15:57

Adding modularity to resource managers and nit changes

d479e91

Consolidated error message

208345c

Uncommenting tests

c9659ce

abhit17 reviewed Jun 21, 2023

View reviewed changes

mmankika-aws added 2 commits June 21, 2023 11:34

Updating paginated call and cleaning up style

7816727

Cleaner implementation of list all resources

84d2f1b

abhit17 previously approved these changes Jun 21, 2023

View reviewed changes

Updating name of few functions

cfbb55b

mmankika-aws dismissed abhit17’s stale review via cfbb55b June 21, 2023 20:51

abhit17 approved these changes Jun 21, 2023

View reviewed changes

mmankika-aws merged commit 53dbb4e into awslabs:master Jun 21, 2023
1 check passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Adding testing architecture and KCL 2.x basic polling/streaming tests #1136

Adding testing architecture and KCL 2.x basic polling/streaming tests #1136

mmankika-aws commented Jun 9, 2023

stair-aws left a comment

stair-aws Jun 13, 2023

mmankika-aws Jun 13, 2023

ZeyuLi-AWS Jun 13, 2023

mmankika-aws Jun 13, 2023

stair-aws left a comment

abhit17 Jun 20, 2023

stair-aws Jun 21, 2023

abhit17 Jun 20, 2023

abhit17 Jun 20, 2023

abhit17 Jun 20, 2023

mmankika-aws Jun 20, 2023

abhit17 Jun 20, 2023

abhit17 Jun 20, 2023

abhit17 Jun 20, 2023

abhit17 Jun 20, 2023

abhit17 Jun 20, 2023

abhit17 Jun 21, 2023

abhit17 Jun 21, 2023

abhit17 Jun 21, 2023

mmankika-aws Jun 21, 2023

abhit17 Jun 21, 2023

abhit17 Jun 21, 2023

abhit17 Jun 21, 2023

abhit17 Jun 21, 2023

abhit17 Jun 21, 2023

mmankika-aws Jun 21, 2023

abhit17 Jun 21, 2023

abhit17 Jun 21, 2023


		public class RecordValidatorQueueTest {

		private RecordValidatorQueue recordValidator = new RecordValidatorQueue();


		RecordValidatorQueue recordValidator;

		public TestRecordProcessorFactory(RecordValidatorQueue recordValidator) {


		private String shardId;

		RecordValidatorQueue recordValidator;

		log.error("Error creating payload data for {}", payloadCounter.toString());
		throw new RuntimeException("Error converting object to bytes: ", e);

		public LeaseTableManager(DynamoDbAsyncClient dynamoClient) throws URISyntaxException, IOException {
		this.dynamoClient = dynamoClient;

Adding testing architecture and KCL 2.x basic polling/streaming tests #1136

Adding testing architecture and KCL 2.x basic polling/streaming tests #1136

Conversation

mmankika-aws commented Jun 9, 2023

stair-aws left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stair-aws left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment