    @Test
    @DisplayName("resend unconfirmed messages")
    void resendUnconfirmedMessagesIntegration() throws Exception {

        Channel ch = connection.createChannel();
        ch.confirmSelect();
        Map<Long, String> outstandingConfirms = resendUnconfirmedMessages(ch);
        assertThat(waitUntil(Duration.ofSeconds(5), () -> outstandingConfirms.isEmpty())).isTrue();
        ch.close();

        Set<String> receivedMessages = ConcurrentHashMap.newKeySet(messageCount);
        CountDownLatch latch = new CountDownLatch(messageCount);
        ch = connection.createChannel();
        ch.basicConsume(queue, true, ((consumerTag, message) -> {
            receivedMessages.add(new String(message.getBody()));
            latch.countDown();
        }), consumerTag -> {
        });

        assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
        assertThat(receivedMessages).hasSize(messageCount);
    }