Allow @Override of KafkaAdmin createAdmin()

[anders-swanson]

Expected Behavior

The Spring KafkaAdmin class has a package-private method "createAdmin" that returns the org.apache.kafka.clients.admin.AdminClient type:

AdminClient createAdmin() {
    return AdminClient.create(this.getAdminConfig());
}

It'd be great if this method was protected, and returned the org.apache.kafka.clients.admin.Admin type. Additionally, the code in KafkaAdmin would need to be updated to use the Admin interface, instead of the KafkaAdmin class.

It's currently possible to do something similar the default producer/consumer factories, as they have methods like this:

    protected Consumer<K, V> createKafkaConsumer(Map<String, Object> configProps) {
        checkBootstrap(configProps);
        Consumer<K, V> kafkaConsumer = createRawConsumer(configProps);
        if (!this.listeners.isEmpty() && !(kafkaConsumer instanceof ExtendedKafkaConsumer)) {
            LOGGER.warn("The 'ConsumerFactory.Listener' configuration is ignored " +
                    "because the consumer is not an instance of 'ExtendedKafkaConsumer'." +
                    "Consider extending 'ExtendedKafkaConsumer' or implement your own 'ConsumerFactory'.");
        }

        for (ConsumerPostProcessor<K, V> pp : this.postProcessors) {
            kafkaConsumer = pp.apply(kafkaConsumer);
        }
        return kafkaConsumer;
    }

Current Behavior

KafkaAdmin cannot be extended/overridden to use another type of class implementing the org.apache.kafka.clients.admin.Admin interface.

Context

I'm using OKafka, which has implementations of the Kafka interfaces, e.g., the Admin interface, and I would like to use these implementations with Spring Kafka. Based on my understanding of the code, some modifications would need to be made as described - I'd also be interested in any workarounds, if they are available.

[sobychacko]

@anders-swanson I think this is a reasonable request. In these cases, we usually recommend providing the correct bootstrap server config to KafkaAdmin as long as you are still using the default AdminClient from the Kafka client. But since you are using a completely different Admin implementation, it makes sense to override that method. We will discuss this further and give an update here. Thanks!

[anders-swanson]

@sobychacko thanks! I created #3484 and ran a few tests against it, seems like it has potential.

[sobychacko]

That's great. We will review it and get back to you.