public <K, V> ConsumerDelegate<K, V> create(LogContext logContext,
                                                Time time,
                                                ConsumerConfig config,
                                                Deserializer<K> keyDeserializer,
                                                Deserializer<V> valueDeserializer,
                                                KafkaClient client,
                                                SubscriptionState subscriptions,
                                                ConsumerMetadata metadata,
                                                List<ConsumerPartitionAssignor> assignors) {
        try {
            GroupProtocol groupProtocol = GroupProtocol.valueOf(config.getString(ConsumerConfig.GROUP_PROTOCOL_CONFIG).toUpperCase(Locale.ROOT));

            if (groupProtocol == GroupProtocol.CONSUMER)
                return new AsyncKafkaConsumer<>(
                    logContext,
                    time,
                    config,
                    keyDeserializer,
                    valueDeserializer,
                    client,
                    subscriptions,
                    metadata
                );
            else
                return new ClassicKafkaConsumer<>(
                    logContext,
                    time,
                    config,
                    keyDeserializer,
                    valueDeserializer,
                    client,
                    subscriptions,
                    metadata,
                    assignors
                );
        } catch (KafkaException e) {
            throw e;
        } catch (Throwable t) {
            throw new KafkaException("Failed to construct Kafka consumer", t);
        }
    }
{@code ConsumerDelegateCreator} implements a quasi-factory pattern to allow the caller to remain unaware of the
underlying {@link Consumer} implementation that is created. This provides the means by which {@link KafkaConsumer}
can remain the top-level facade for implementations, but allow different implementations to co-exist under
the covers.

<p/>

The current logic for the {@code ConsumerCreator} inspects the incoming configuration and determines if
it is using the new consumer group protocol (KIP-848) or if it should fall back to the existing, legacy group
protocol. This is based on the presence and value of the {@link ConsumerConfig#GROUP_PROTOCOL_CONFIG group.protocol}
configuration. If the value is present and equal to &quot;{@code consumer}&quot;, the {@link AsyncKafkaConsumer}
will be returned. Otherwise, the {@link ClassicKafkaConsumer} will be returned.


<p/>

<em>Note</em>: this is for internal use only and is not intended for use by end users. Internal users should
not attempt to determine the underlying implementation to avoid coding to an unstable interface. Rather, it is
the {@link Consumer} API contract that should serve as the caller's interface.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerDelegateCreator.java