public StreamsResponse getKinesisStreamNames(AWSRequest request) throws ExecutionException {

        LOG.debug("List Kinesis streams for region [{}]", request.region());

        final KinesisClient kinesisClient = awsClientBuilderUtil.buildClient(kinesisClientBuilder, request);

        ListStreamsRequest streamsRequest = ListStreamsRequest.builder().limit(KINESIS_LIST_STREAMS_LIMIT).build();
        final ListStreamsResponse listStreamsResponse = kinesisClient.listStreams(streamsRequest);
        final List<String> streamNames = new ArrayList<>(listStreamsResponse.streamNames());

        final Retryer<Boolean> retryer = RetryerBuilder.<Boolean>newBuilder()
                .retryIfResult(b -> Objects.equals(b, Boolean.TRUE))
                .retryIfExceptionOfType(LimitExceededException.class)
                .withStopStrategy(StopStrategies.stopAfterAttempt(KINESIS_LIST_STREAMS_MAX_ATTEMPTS))
                .build();

        if (listStreamsResponse.hasMoreStreams()) {
            try {
                retryer.call(() -> {
                    LOG.debug("Requesting streams...");
                    final String lastStreamName = streamNames.get(streamNames.size() - 1);
                    final ListStreamsRequest moreStreamsRequest = ListStreamsRequest.builder()
                            .exclusiveStartStreamName(lastStreamName)
                            .limit(KINESIS_LIST_STREAMS_LIMIT).build();
                    final ListStreamsResponse moreSteamsResponse = kinesisClient.listStreams(moreStreamsRequest);
                    streamNames.addAll(moreSteamsResponse.streamNames());

                    // If more streams, then this will execute again.
                    return moreSteamsResponse.hasMoreStreams();
                });
            } catch (RetryException e) {
                LOG.error("Failed to get all stream names after {} attempts. Proceeding to return currently obtained streams.", KINESIS_LIST_STREAMS_MAX_ATTEMPTS);
            }
        }
        LOG.debug("Kinesis streams queried: [{}]", streamNames);

        if (streamNames.isEmpty()) {
            throw new BadRequestException(String.format(Locale.ROOT, "No Kinesis streams were found in the [%s] region.", request.region()));
        }

        return StreamsResponse.create(streamNames, streamNames.size());
    }
@Test
    public void testGetStreams() throws ExecutionException {

        when(awsClientBuilderUtil.buildClient(any(KinesisClientBuilder.class), any())).thenReturn(kinesisClient);

        // Test with two streams and one page. This is the most common case for most AWS accounts.
        when(kinesisClient.listStreams(isA(ListStreamsRequest.class)))
                .thenReturn(ListStreamsResponse.builder()
                        .streamNames(TWO_TEST_STREAMS)
                        .hasMoreStreams(false).build());

        StreamsResponse streamsResponse = kinesisService.getKinesisStreamNames(AWSRequestImpl.builder()
                .region(TEST_REGION)
                .awsAccessKeyId("a-key")
                .awsSecretAccessKey(encryptedValue).build());
        assertEquals(2, streamsResponse.total());
        assertEquals(2, streamsResponse.streams().size());

        // Test with stream paging functionality. This will be the case when a large number of Kinesis streams
        // are present on a particular AWS account.
        when(kinesisClient.listStreams(isA(ListStreamsRequest.class)))
                // First return a response with two streams indicating that there are more.
                .thenReturn(ListStreamsResponse.builder()
                        .streamNames(TWO_TEST_STREAMS)
                        .hasMoreStreams(true).build())
                // Then return a response with two streams and indicate that all have been retrieved.
                .thenReturn(ListStreamsResponse.builder()
                        .streamNames(TWO_TEST_STREAMS)
                        .hasMoreStreams(false).build()); // Indicate no more streams.

        streamsResponse = kinesisService.getKinesisStreamNames(AWSRequestImpl.builder()
                .region(TEST_REGION)
                .awsAccessKeyId("a-key")
                .awsSecretAccessKey(encryptedValue).build());

        // There should be 4 total streams (two from each page).
        assertEquals(4, streamsResponse.total());
        assertEquals(4, streamsResponse.streams().size());
    }