@Override
    protected int poll() throws Exception {
        // must reset for each poll
        shutdownRunningTask = null;
        pendingExchanges = 0;

        List<software.amazon.awssdk.services.sqs.model.Message> messages = pollingTask.call();
        // okay we have some response from aws so lets mark the consumer as ready
        forceConsumerAsReady();

        Queue<Exchange> exchanges = createExchanges(messages);
        return processBatch(CastUtils.cast(exchanges));
    }
@Test
    void shouldAutomaticallyCreateQueueOnQueueDoesNotExistException() throws Exception {
        // given
        configuration.setAutoCreateQueue(true);
        sqsClientMock.setReceiveRequestHandler(request -> {
            throw QueueDoesNotExistException.builder().build();
        });
        sqsClientMock.setQueueUrl(null);
        try (var tested = createConsumer(5)) {

            // when
            var polledMessagesCount = tested.poll();

            // then
            assertThat(polledMessagesCount).isZero();
            assertThat(receivedExchanges).isEmpty();
            assertThat(sqsClientMock.getReceiveRequests()).containsExactlyInAnyOrder(expectedReceiveRequest(5));
            assertThat(sqsClientMock.getQueueUrlRequests()).containsExactly(GetQueueUrlRequest.builder()
                    .queueName(configuration.getQueueName())
                    .build());
            assertThat(sqsClientMock.getCreateQueueRequets()).containsExactly(CreateQueueRequest.builder()
                    .queueName(configuration.getQueueName())
                    .attributes(emptyMap())
                    .build());
        }
    }