static boolean waitUntil(Duration timeout, BooleanSupplier condition) throws InterruptedException {
        int waited = 0;
        while (!condition.getAsBoolean() && waited < timeout.toMillis()) {
            Thread.sleep(100L);
            waited += 100;
        }
        return condition.getAsBoolean();
    }
@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);
    }