add simple tests for Kinesis consumers using v1 and v2 SDKs (#28)

Author: pinzon

src/test/java/cloud/localstack/KinesisConsumerTest.java

@@ -0,0 +1,83 @@
+package cloud.localstack.awssdkv1;
+
+import cloud.localstack.LocalstackTestRunner;
+
+import com.amazonaws.services.kinesis.AmazonKinesisAsync;
+import com.amazonaws.services.kinesis.model.CreateStreamRequest;
+import com.amazonaws.services.kinesis.model.PutRecordRequest;
+import com.amazonaws.services.kinesis.model.PutRecordResult;
+import com.amazonaws.services.kinesis.model.GetRecordsRequest;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.internal.SdkInternalList;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import static java.lang.System.out;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+import java.nio.ByteBuffer;
+
+@RunWith(LocalstackTestRunner.class)
+public class KinesisConsumerTest {
+
+    @Test
+    public void testGetRecord() throws Exception{
+      String streamName = "test-s-"+UUID.randomUUID().toString();
+      AmazonKinesisAsync kinesisClient = TestUtils.getClientKinesisAsync();
+
+      CreateStreamRequest createStreamRequest = new CreateStreamRequest();
+      createStreamRequest.setStreamName(streamName);
+      createStreamRequest.setShardCount(1);
+
+      kinesisClient.createStream(createStreamRequest);
+      TimeUnit.SECONDS.sleep(2);
+
+      PutRecordRequest putRecordRequest = new PutRecordRequest();
+      putRecordRequest.setPartitionKey("partitionkey");
+      putRecordRequest.setStreamName(streamName);
+      putRecordRequest.setData(ByteBuffer.wrap("hello, world!".getBytes()));
+
+      String shardId = kinesisClient.putRecord(putRecordRequest).getShardId();
+
+      GetShardIteratorRequest getShardIteratorRequest = new GetShardIteratorRequest();
+      getShardIteratorRequest.setShardId(shardId);
+      getShardIteratorRequest.setShardIteratorType("TRIM_HORIZON");
+      getShardIteratorRequest.setStreamName(streamName);
+
+      String shardIterator = kinesisClient
+        .getShardIterator(getShardIteratorRequest)
+        .getShardIterator();
+
+      GetRecordsRequest getRecordRequest = new GetRecordsRequest() ;
+      getRecordRequest.setShardIterator(shardIterator);
+
+      Integer limit = 100;
+      Integer counter = 0;
+      Boolean recordFound = false; 
+      
+      while (true) {
+        getRecordRequest.setShardIterator(shardIterator);
+        GetRecordsResult recordsResponse = kinesisClient.getRecords(getRecordRequest);
+        
+        List records = recordsResponse.getRecords();
+        if (records.isEmpty()) {
+          recordFound = true; 
+          break;
+        }
+
+        if(counter >= limit){
+          break;
+        }
+
+        counter += 1;
+        shardIterator = recordsResponse.getNextShardIterator();
+      }
+      Assert.assertTrue(recordFound);
+    }
+}

src/test/java/cloud/localstack/awssdkv2/KinesisV2ConsumerTest.java

@@ -0,0 +1,85 @@
+package cloud.localstack.awssdkv2;
+
+import cloud.localstack.LocalstackTestRunner;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.kinesis.model.CreateStreamRequest;
+import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
+import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
+import software.amazon.awssdk.http.Protocol;
+import software.amazon.awssdk.utils.AttributeMap;
+import software.amazon.awssdk.http.nio.netty.*;
+import software.amazon.awssdk.core.*;
+import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
+import software.amazon.awssdk.services.kinesis.model.*;
+
+import static java.lang.System.out;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+@RunWith(LocalstackTestRunner.class)
+public class KinesisV2ConsumerTest {
+
+    @Test
+    public void testGetRecord() throws Exception{
+      String streamName = "test-s-"+UUID.randomUUID().toString();
+      KinesisAsyncClient kinesisClient = TestUtils.getClientKinesisAsyncV2();
+
+      CreateStreamRequest request = CreateStreamRequest.builder()
+          .streamName(streamName).shardCount(1).build();
+      CreateStreamResponse response = kinesisClient.createStream(request).get();
+      Assert.assertNotNull(response);
+      TimeUnit.SECONDS.sleep(2);
+
+      PutRecordRequest putRecordRequest = PutRecordRequest.builder()
+        .partitionKey("partitionkey")
+        .streamName(streamName)
+        .data(SdkBytes.fromUtf8String("hello, world!"))
+        .build();
+      String shardId = kinesisClient.putRecord(putRecordRequest).get().shardId();
+
+      GetShardIteratorRequest getShardIteratorRequest = GetShardIteratorRequest.builder()
+        .shardId(shardId)
+        .shardIteratorType(ShardIteratorType.TRIM_HORIZON)
+        .streamName(streamName)
+        .build();
+      String shardIterator = kinesisClient
+        .getShardIterator(getShardIteratorRequest)
+        .get()
+        .shardIterator();
+
+      GetRecordsRequest getRecordRequest = GetRecordsRequest.builder().shardIterator(shardIterator).build();
+      Integer limit = 100;
+      Integer counter = 0;
+      Boolean recordFound = false;
+      
+      while (true) {
+        GetRecordsResponse recordsResponse = kinesisClient.getRecords(getRecordRequest).get();
+        
+        if (recordsResponse.hasRecords()) {
+          recordFound = true;
+          break;
+        }
+
+        if(counter >= limit){
+          break;
+        }
+
+        counter += 1;
+        shardIterator = recordsResponse.nextShardIterator();
+      }
+      Assert.assertTrue(recordFound);
+    }
+}