public abstract List<Consumer<T>> getInputConsumers();
@Test(dataProvider = "sourceImpls")
    public void testInputConsumersGetter(PulsarSourceConfig pulsarSourceConfig) throws Exception {
        PulsarSource<GenericRecord> pulsarSource = getPulsarSource(pulsarSourceConfig);
        pulsarSource.open(new HashMap<>(), null);

        if (pulsarSourceConfig instanceof SingleConsumerPulsarSourceConfig) {
            SingleConsumerPulsarSourceConfig cfg = (SingleConsumerPulsarSourceConfig) pulsarSourceConfig;
            Assert.assertEquals(pulsarSource.getInputConsumers().size(), 1);
            return;
        }

        if (pulsarSourceConfig instanceof MultiConsumerPulsarSourceConfig) {
            MultiConsumerPulsarSourceConfig cfg = (MultiConsumerPulsarSourceConfig) pulsarSourceConfig;
            Assert.assertEquals(pulsarSource.getInputConsumers().size(), cfg.getTopicSchema().size());
            return;
        }

        fail("Unknown config type");
    }