Fix flaky KinesisStabilityTest

Author: zoewangg

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

@@ -183,8 +183,10 @@ private List<CompletableFuture<?>> generateSubscribeToShardFutures() {
                                                        .startingPosition(s -> s.type(ShardIteratorType.TRIM_HORIZON)),
                                                  responseHandler)
                                .thenAccept(b -> {
-                                   // Only verify data if all events have been received.
-                                   if (responseHandler.allEventsReceived) {
+                                   // 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);
                                    }
@@ -218,7 +220,7 @@ private static class TestSubscribeToShardResponseHandler extends TestEventStream
         , SubscribeToShardEventStream> implements SubscribeToShardResponseHandler {
         private final List<SdkBytes> receivedData = new ArrayList<>();
         private final String id;
-        private boolean allEventsReceived = false;
+        private volatile boolean allEventsReceived = false;
 
         TestSubscribeToShardResponseHandler(int consumerIndex, int shardIndex) {
             id = "consumer_" + consumerIndex + "_shard_" + shardIndex;