public static Queue<Consumer<byte[]>> stopConsumers(final Queue<Consumer<byte[]>> consumers) throws PulsarClientException {
        while (!consumers.isEmpty()) {
            Consumer<byte[]> consumer = consumers.poll();
            if (consumer != null) {
                try {
                    consumer.close();
                } catch (PulsarClientException.AlreadyClosedException e) {
                    // ignore during stopping
                } catch (Exception e) {
                    LOG.debug("Error stopping consumer: {} due to {}. This exception is ignored", consumer,
                            e.getMessage(), e);
                }
            }
        }

        return new ConcurrentLinkedQueue<>();
    }
@Test
    public void givenConsumerQueueIsNotEmptywhenIStopConsumersverifyEmptyQueueIsReturned() throws PulsarClientException {
        Queue<Consumer<byte[]>> consumers = new ConcurrentLinkedQueue<>();
        consumers.add(mock(Consumer.class));

        Queue<Consumer<byte[]>> expected = PulsarUtils.stopConsumers(consumers);

        assertTrue(expected.isEmpty());
    }