public KinesisNewStreamResponse createNewKinesisStream(KinesisNewStreamRequest request) {
        LOG.debug("Creating Kinesis client with the provided credentials.");
        final KinesisClient kinesisClient = awsClientBuilderUtil.buildClient(kinesisClientBuilder, request);

        LOG.debug("Creating new Kinesis stream request [{}].", request.streamName());
        final CreateStreamRequest createStreamRequest = CreateStreamRequest.builder()
                .streamName(request.streamName())
                .shardCount(SHARD_COUNT)
                .build();
        LOG.debug("Sending request to create new Kinesis stream [{}] with [{}] shards.",
                request.streamName(), SHARD_COUNT);

        StreamDescription streamDescription;
        try {
            kinesisClient.createStream(createStreamRequest);
            int seconds = 0;
            do {
                try {
                    Thread.sleep(1_000);
                } catch (InterruptedException e) {
                    LOG.error("Request interrupted while waiting for shard to become available.");
                    return null; // Give up on request.
                }
                streamDescription = kinesisClient
                        .describeStream(DescribeStreamRequest.builder().streamName(request.streamName()).build())
                        .streamDescription();
                if (seconds > 300) {
                    final String responseMessage = String.format(Locale.ROOT, "Fail. Stream [%s] has failed to become active " +
                            "within 60 seconds.", request.streamName());
                    throw new BadRequestException(responseMessage);
                }
                seconds++;
            } while (streamDescription.streamStatus() != StreamStatus.ACTIVE);
            String streamArn = streamDescription.streamARN();
            final String responseMessage = String.format(Locale.ROOT, "Success. The new stream [%s/%s] was created with [%d] shard.",
                    request.streamName(), streamArn, SHARD_COUNT);

            return KinesisNewStreamResponse.create(createStreamRequest.streamName(),
                    streamArn,
                    responseMessage);
        } catch (Exception e) {
            final String specificError = ExceptionUtils.formatMessageCause(e);
            final String responseMessage = String.format(Locale.ROOT, "Attempt to create [%s] new Kinesis stream " +
                            "with [%d] shards failed due to the following exception: [%s]",
                    request.streamName(), SHARD_COUNT,
                    specificError);
            LOG.error(responseMessage, e);
            throw new BadRequestException(responseMessage, e);
        }
    }
@Test
    public void testCreateNewKinesisStream() {

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

        // Mock out specific KinesisNewStreamRequest to return a response.
        when(kinesisClient.createStream(isA(CreateStreamRequest.class))).thenReturn(CreateStreamResponse.builder().build());
        when(kinesisClient.describeStream(isA(DescribeStreamRequest.class)))
                .thenReturn(DescribeStreamResponse.builder().streamDescription(StreamDescription.builder()
                        .streamName(TEST_STREAM_1)
                        .streamStatus(StreamStatus.ACTIVE)
                        .streamARN(STREAM_ARN)
                        .build()).build());

        final KinesisNewStreamRequest kinesisNewStreamRequest = KinesisNewStreamRequest.builder()
                .region(Region.EU_WEST_1.id())
                .awsAccessKeyId("a-key")
                .awsSecretAccessKey(encryptedValue)
                .streamName(TEST_STREAM_1).build();
        // TODO debug the error
        final KinesisNewStreamResponse response = kinesisService.createNewKinesisStream(kinesisNewStreamRequest);

        // Check the values are whats expected.
        final String expectedResponse = "Success. The new stream [" + TEST_STREAM_1 + "/" + STREAM_ARN + "] was created with [1] shard.";
        assertEquals(expectedResponse, response.result());
        assertEquals(SHARD_COUNT, 1);
    }