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 givenConsumerQueueIsNotEmptywhenIStopConsumersverifyCallToCloseAndUnsubscribeConsumer()
            throws PulsarClientException {
        Consumer<byte[]> consumer = mock(Consumer.class);

        Queue<Consumer<byte[]>> consumers = new ConcurrentLinkedQueue<>();
        consumers.add(consumer);

        PulsarUtils.stopConsumers(consumers);

        verify(consumer).close();
    }