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("allow max of outstanding confirms")
    void allowMaxOfOutstandingConfirms() throws Exception {
        int maxOutstandingConfirms = 100;
        Semaphore semaphore = new Semaphore(maxOutstandingConfirms);
        Channel ch = connection.createChannel();
        ch.confirmSelect();
        ConcurrentSkipListSet<Long> outstandingConfirms = new ConcurrentSkipListSet<>();
        ConfirmCallback handleConfirm = (deliveryTag, multiple) -> {
            if (multiple) {
                NavigableSet<Long> confirmed = outstandingConfirms.headSet(deliveryTag, true);
                int confirmedSize = confirmed.size();
                confirmed.clear();
                semaphore.release(confirmedSize);
            } else {
                outstandingConfirms.remove(deliveryTag);
                semaphore.release();
            }
        };
        ch.addConfirmListener(handleConfirm, handleConfirm);
        for (int i = 0; i < messageCount; i++) {
            String body = String.valueOf(i);
            boolean acquired = semaphore.tryAcquire(10, TimeUnit.SECONDS);
            if (!acquired) {
                throw new IllegalStateException("Could not publish because of too many outstanding publisher confirms");
            }
            outstandingConfirms.add(ch.getNextPublishSeqNo());
            ch.basicPublish("", queue, null, body.getBytes());
            meter.mark();
        }
        assertThat(waitUntil(Duration.ofSeconds(5), () -> outstandingConfirms.isEmpty())).isTrue();
        ch.close();

        CountDownLatch latch = new CountDownLatch(messageCount);
        ch = connection.createChannel();
        ch.basicConsume(queue, true, ((consumerTag, message) -> latch.countDown()), consumerTag -> {
        });

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