(self)
return NewKafkaConsumerCheck(self)
Determine whether to use old legacy KafkaClient implementation or the new KafkaAdminClient implementation.

        The legacy version of this check uses the KafkaClient and handrolls things like looking up the GroupCoordinator,
        crafting the offset requests, handling errors, etc.

        The new implementation uses the KafkaAdminClient which lets us offload most of the Kafka-specific bits onto the
        kafka-python library, which is used by many other tools and reduces our maintenance burden.

        Unfortunately, the KafkaAdminClient requires brokers >= 0.10.0, so we split the check into legacy and new code.

        Furthermore, we took the opportunity to simplify the new code by dropping support for:
        1) Zookeeper-based offsets. These have been deprecated since Kafka 0.9.
        2) Kafka brokers < 0.10.2. It is impossible to support monitor_unlisted_consumer_groups on these older brokers
        because they do not provide a way to determine the mapping of consumer groups to topics. For details, see
        KIP-88.

        To clarify: This check still allows fetching offsets from zookeeper/older kafka brokers, it just uses the
        legacy code path.
Determine whether to use old legacy KafkaClient implementation or the new KafkaAdminClient implementation.
Determine whether to use old legacy KafkaClient implementation or the new KafkaAdminClient implementation .
def _make_sub_check(self):
        """Determine whether to use old legacy KafkaClient implementation or the new KafkaAdminClient implementation.

        The legacy version of this check uses the KafkaClient and handrolls things like looking up the GroupCoordinator,
        crafting the offset requests, handling errors, etc.

        The new implementation uses the KafkaAdminClient which lets us offload most of the Kafka-specific bits onto the
        kafka-python library, which is used by many other tools and reduces our maintenance burden.

        Unfortunately, the KafkaAdminClient requires brokers >= 0.10.0, so we split the check into legacy and new code.

        Furthermore, we took the opportunity to simplify the new code by dropping support for:
        1) Zookeeper-based offsets. These have been deprecated since Kafka 0.9.
        2) Kafka brokers < 0.10.2. It is impossible to support monitor_unlisted_consumer_groups on these older brokers
        because they do not provide a way to determine the mapping of consumer groups to topics. For details, see
        KIP-88.

        To clarify: This check still allows fetching offsets from zookeeper/older kafka brokers, it just uses the
        legacy code path."""
        if self.instance.get('zk_connect_str'):
            return LegacyKafkaCheck_0_10_2(self)

        kafka_version = self.instance.get('kafka_client_api_version')
        if isinstance(kafka_version, str):
            kafka_version = tuple(map(int, kafka_version.split(".")))
        if kafka_version is None:  # if unspecified by the user, we have to probe the cluster
            kafka_client = self.create_kafka_client()
            # version probing happens automatically as part of KafkaClient's __init__()
            kafka_version = kafka_client.config['api_version']
            kafka_client.close()

        if kafka_version < (0, 10, 2):
            return LegacyKafkaCheck_0_10_2(self)

        return NewKafkaConsumerCheck(self)
def _make_sub_check ( self ) : if self . instance . get ( 'zk_connect_str' ) : return LegacyKafkaCheck_0_10_2 ( self ) kafka_version = self . instance . get ( 'kafka_client_api_version' ) if isinstance ( kafka_version , str ) : kafka_version = tuple ( map ( int , kafka_version . split ( "." ) ) ) if kafka_version is None : # if unspecified by the user, we have to probe the cluster kafka_client = self . create_kafka_client ( ) # version probing happens automatically as part of KafkaClient's __init__() kafka_version = kafka_client . config [ 'api_version' ] kafka_client . close ( ) if kafka_version < ( 0 , 10 , 2 ) : return LegacyKafkaCheck_0_10_2 ( self ) return NewKafkaConsumerCheck ( self )