Add backoff with jitter to Kinesis stabiltiy test

Author: RanVaknin

test/stability-tests/src/it/java/software/amazon/awssdk/stability/tests/kinesis/KinesisStabilityTest.java

@@ -31,6 +31,7 @@
 import org.apache.commons.lang3.RandomUtils;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.RepeatedTest;
 import software.amazon.awssdk.core.SdkBytes;
 import software.amazon.awssdk.core.async.SdkPublisher;
 import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
@@ -115,16 +116,17 @@ public void tearDown() {
         asyncClient.close();
     }
 
+    @RepeatedTest(30)
     @RetryableTest(maxRetries = 3, retryableException = StabilityTestsRetryableException.class)
-    public void putRecords_subscribeToShard() {
+    public void putRecords_subscribeToShard() throws InterruptedException {
         putRecords();
         subscribeToShard();
     }
 
     /**
      * We only have one run of subscribeToShard tests because it takes 5 minutes.
      */
-    private void subscribeToShard() {
+    private void subscribeToShard() throws InterruptedException {
         log.info(() -> "starting to test subscribeToShard to stream: " + streamName);
         List<CompletableFuture<?>> completableFutures = generateSubscribeToShardFutures();
         StabilityTestRunner.newRunner()
@@ -170,12 +172,15 @@ private void putRecords() {
      * Generate request per consumer/shard combination
      * @return a lit of completablefutures
      */
-    private List<CompletableFuture<?>> generateSubscribeToShardFutures() {
+    private List<CompletableFuture<?>> generateSubscribeToShardFutures() throws InterruptedException {
         List<CompletableFuture<?>> completableFutures = new ArrayList<>();
+        int baseDelay = 150;
+        int jitterRange = 150;
         for (int i = 0; i < CONSUMER_COUNT; i++) {
             final int consumerIndex = i;
             for (int j = 0; j < SHARD_COUNT; j++) {
                 final int shardIndex = j;
+                Thread.sleep(baseDelay + (int)(Math.random() * jitterRange));
                 TestSubscribeToShardResponseHandler responseHandler =
                     new TestSubscribeToShardResponseHandler(consumerIndex, shardIndex);
                 CompletableFuture<Void> completableFuture =
@@ -184,9 +189,6 @@ private List<CompletableFuture<?>> generateSubscribeToShardFutures() {
                                                        .startingPosition(s -> s.type(ShardIteratorType.TRIM_HORIZON)),
                                                  responseHandler)
                                .thenAccept(b -> {
-                                   // Only verify data if all events have been received and the received data is not empty.
-                                   // It is possible the received data is empty because there is no record at the position
-                                   // event with TRIM_HORIZON.
                                    if (responseHandler.allEventsReceived && !responseHandler.receivedData.isEmpty()) {
                                        assertThat(producedData).as(responseHandler.id + " has not received all events"
                                                                    + ".").containsSequence(responseHandler.receivedData);