public List<ConsumerImpl<T>> getConsumers() {
        return consumers.values().stream().collect(Collectors.toList());
    }
@Test
    public void testConsumerCleanupOnSubscribeFailure() {
        ConsumerConfigurationData<byte[]> consumerConfData = new ConsumerConfigurationData<>();
        consumerConfData.setSubscriptionName("subscriptionName");
        consumerConfData.setTopicNames(new HashSet<>(Arrays.asList("a", "b", "c")));
        int completionDelayMillis = 10;
        Schema<byte[]> schema = Schema.BYTES;
        PulsarClientImpl clientMock = createPulsarClientMockWithMockedClientCnx(executorProvider, internalExecutor);
        when(clientMock.getPartitionedTopicMetadata(any(), anyBoolean(), anyBoolean()))
                .thenAnswer(invocation -> createExceptionFuture(
                new PulsarClientException.InvalidConfigurationException("a mock exception"), completionDelayMillis));
        CompletableFuture<Consumer<byte[]>> completeFuture = new CompletableFuture<>();
        MultiTopicsConsumerImpl<byte[]> impl = new MultiTopicsConsumerImpl<byte[]>(clientMock, consumerConfData,
                executorProvider, completeFuture, schema, null, true);
        // assert that we don't start in closed, then we move to closed and get an exception
        // indicating that closeAsync was called
        assertEquals(impl.getState(), HandlerState.State.Uninitialized);
        try {
            completeFuture.get(2, TimeUnit.SECONDS);
        } catch (Throwable ignore) {
            // just ignore the exception
        }
        assertTrue(completeFuture.isCompletedExceptionally());
        assertEquals(impl.getConsumers().size(), 0);
        assertEquals(impl.getState(), HandlerState.State.Closed);
        verify(clientMock, times(1)).cleanupConsumer(any());
    }