public abstract List<Consumer<T>> getInputConsumers();
@Test
    public void testDanglingSubscriptions() throws Exception {
        MultiConsumerPulsarSourceConfig pulsarConfig = getMultiConsumerPulsarConfigs(true);

        MultiConsumerPulsarSource<?> pulsarSource =
                new MultiConsumerPulsarSource<>(getPulsarClient(), pulsarConfig, new HashMap<>(),
                        Thread.currentThread().getContextClassLoader());
        try {
            pulsarSource.open(new HashMap<>(), Mockito.mock(SourceContext.class));
            fail();
        } catch (CompletionException e) {
            pulsarSource.close();
            assertEquals(pulsarSource.getInputConsumers().size(), 1);
        } catch (Exception e) {
            fail();
        }

    }