tests(kinesis): attempt to fix flaky tests (#670)

Author: Malaydewangan09

src/test/java/io/kestra/plugin/aws/kinesis/AbstractKinesisTest.java

@@ -23,36 +23,36 @@ public class AbstractKinesisTest extends AbstractLocalStackTest {
     @Inject
     protected RunContextFactory runContextFactory;
     protected static String streamArn;
-    protected static final String STREAM_NAME = "stream";
+    protected static String streamName;
 
     @BeforeAll
-    static void beforeAll() throws InterruptedException {
-        try(KinesisClient kinesisClient = KinesisClient.builder()
-            .credentialsProvider(StaticCredentialsProvider.create(AwsBasicCredentials.create(
-                localstack.getAccessKey(),
-                localstack.getSecretKey()
-            )))
+    static void setupStream() throws InterruptedException {
+        streamName = "stream-" + IdUtils.create();
+
+        try (KinesisClient kinesisClient = KinesisClient.builder()
+            .credentialsProvider(StaticCredentialsProvider.create(
+                AwsBasicCredentials.create(localstack.getAccessKey(), localstack.getSecretKey())
+            ))
             .region(Region.of(localstack.getRegion()))
             .endpointOverride(localstack.getEndpoint())
             .build()) {
 
-            try {
-                kinesisClient.createStream(CreateStreamRequest.builder()
-                    .streamName(STREAM_NAME)
-                    .shardCount(1)
-                    .build());
-            } catch (ResourceInUseException ignored) {}
+            kinesisClient.createStream(CreateStreamRequest.builder()
+                .streamName(streamName)
+                .shardCount(1)
+                .build());
 
-            DescribeStreamResponse ds = kinesisClient.describeStream(r -> r.streamName(STREAM_NAME));
-            while (ds.streamDescription().streamStatus() != StreamStatus.ACTIVE) {
+            DescribeStreamResponse ds;
+            do {
                 Thread.sleep(200);
-                ds = kinesisClient.describeStream(r -> r.streamName(STREAM_NAME));
-            }
+                ds = kinesisClient.describeStream(r -> r.streamName(streamName));
+            } while (ds.streamDescription().streamStatus() != StreamStatus.ACTIVE);
 
             streamArn = ds.streamDescription().streamARN();
         }
     }
 
+
     protected String registerConsumer() throws Exception {
         try (KinesisClient kinesis = KinesisClient.builder()
             .credentialsProvider(StaticCredentialsProvider.create(AwsBasicCredentials.create(

src/test/java/io/kestra/plugin/aws/kinesis/ConsumeTest.java

@@ -37,7 +37,7 @@ void testConsume() throws Exception {
             .region(Property.ofValue(localstack.getRegion()))
             .accessKeyId(Property.ofValue(localstack.getAccessKey()))
             .secretKeyId(Property.ofValue(localstack.getSecretKey()))
-            .streamName(Property.ofValue("stream"))
+            .streamName(Property.ofValue(streamName))
             .records(List.of(record))
             .build();
 
@@ -48,7 +48,7 @@ void testConsume() throws Exception {
             .region(Property.ofValue(localstack.getRegion()))
             .accessKeyId(Property.ofValue(localstack.getAccessKey()))
             .secretKeyId(Property.ofValue(localstack.getSecretKey()))
-            .streamName(Property.ofValue("stream"))
+            .streamName(Property.ofValue(streamName))
             .iteratorType(Property.ofValue(AbstractKinesis.IteratorType.TRIM_HORIZON))
             .maxRecords(Property.ofValue(10))
             .pollDuration(Property.ofValue(java.time.Duration.ofSeconds(1)))

src/test/java/io/kestra/plugin/aws/kinesis/RealtimeTriggerTest.java

@@ -67,15 +67,15 @@ void evaluate() throws Exception {
                 triggers:
                   - id: realtime
                     type: io.kestra.plugin.aws.kinesis.RealtimeTrigger
-                    streamName: "stream"
+                    streamName: "%s"
                     consumerArn: "%s"
                     region: "us-east-1"
                     accessKeyId: "test"
                     secretKeyId: "test"
                     endpointOverride: "http://localhost:4566"
                     iteratorType: TRIM_HORIZON
                 """
-                .formatted(consumerArn);
+                .formatted(streamName, consumerArn);
 
 
             File tempFlow = File.createTempFile("kinesis-realtime", ".yaml");
@@ -93,7 +93,7 @@ void evaluate() throws Exception {
                 .region(Property.ofValue(localstack.getRegion()))
                 .accessKeyId(Property.ofValue(localstack.getAccessKey()))
                 .secretKeyId(Property.ofValue(localstack.getSecretKey()))
-                .streamName(Property.ofValue("stream"))
+                .streamName(Property.ofValue(streamName))
                 .records(List.of(record))
                 .build();
 

src/test/java/io/kestra/plugin/aws/kinesis/TriggerTest.java

@@ -41,7 +41,7 @@ void evaluate() throws Exception {
             .region(Property.ofValue(localstack.getRegion()))
             .accessKeyId(Property.ofValue(localstack.getAccessKey()))
             .secretKeyId(Property.ofValue(localstack.getSecretKey()))
-            .streamName(Property.ofValue("stream"))
+            .streamName(Property.ofValue(streamName))
             .records(List.of(record, record2, record3))
             .build();
 
@@ -50,7 +50,7 @@ void evaluate() throws Exception {
         var trigger = Trigger.builder()
             .id(TriggerTest.class.getSimpleName())
             .type(TriggerTest.class.getName())
-            .streamName(Property.ofValue("stream"))
+            .streamName(Property.ofValue(streamName))
             .iteratorType(Property.ofValue(AbstractKinesis.IteratorType.TRIM_HORIZON))
             .maxRecords(Property.ofValue(3))
             .region(Property.ofValue(localstack.getRegion()))