@VisibleForTesting
    Consumer<?> getConsumer(String topic, int partition) throws PulsarClientException {
        if (inputConsumers == null) {
            throw new PulsarClientException("Getting consumer is not supported");
        }

        Consumer<?> consumer = tryGetConsumer(topic, partition);
        if (consumer == null) {
            // MultiTopicsConsumer's list of consumers could change
            // if partitions changed or pattern(s) used to subscribe.
            // Reload and try one more time.
            reloadConsumersFromMultiTopicsConsumers();
            consumer = tryGetConsumer(topic, partition);
        }

        if (consumer != null) {
            return consumer;
        }
        throw new PulsarClientException("Consumer for topic " + topic
                + " partition " + partition + " is not found");
    }
@Test
    public void testGetConsumer() throws PulsarClientException {
        config.setExposePulsarAdminClientEnabled(false);
        context = new ContextImpl(
                config,
                logger,
                client,
                new EnvironmentBasedSecretsProvider(), FunctionCollectorRegistry.getDefaultImplementation(),
                new String[0],
                FunctionDetails.ComponentType.FUNCTION, null, new InstanceStateManager(),
                pulsarAdmin, clientBuilder, t -> {}, producerCache);
        Consumer<?> mockConsumer = Mockito.mock(Consumer.class);
        when(mockConsumer.getTopic()).thenReturn(TopicName.get("z").toString());
        context.setInputConsumers(Lists.newArrayList(mockConsumer));

        Assert.assertNotNull(context.getConsumer("z", 0));
        try {
            context.getConsumer("z", 1);
            Assert.fail("Expected exception");
        } catch (PulsarClientException e) {
            // pass
        }
    }