<s> """<STR_LIT>""" <EOL> import json <EOL> import logging <EOL> import platform <EOL> import random <EOL> import time <EOL> import uuid <EOL> from collections import namedtuple , Mapping , defaultdict <EOL> from itertools import chain <EOL> import zookeeper <EOL> from zc . zk import ZooKeeper , FailedConnect <EOL> from brod . base import BrokerPartition , ConsumerStats , MessageSet <EOL> from brod . base import ConnectionFailure , FetchResult , KafkaError , OffsetOutOfRange <EOL> from brod . simple import SimpleConsumer <EOL> from brod . blocking import Kafka <EOL> log = logging . getLogger ( '<STR_LIT>' ) <EOL> class NoAvailablePartitionsError ( KafkaError ) : pass <EOL> class ConsumerEntryNotFoundError ( KafkaError ) : pass <EOL> class ZKConnectError ( KafkaError ) : pass <EOL> class ZKUtil ( object ) : <EOL> ACL = [ { "<STR_LIT>" : <NUM_LIT> , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:id>" : "<STR_LIT>" } ] <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , zk_conn_str , zk_timeout = None ) : <EOL> try : <EOL> self . _zk = ZooKeeper ( zk_conn_str , zk_timeout ) <EOL> except FailedConnect as e : <EOL> raise ZKConnectError ( e ) <EOL> def close ( self ) : <EOL> self . _zk . close ( ) <EOL> def broker_partitions_for ( self , topic , force_partition_zero = False ) : <EOL> """<STR_LIT>""" <EOL> if force_partition_zero : <EOL> broker_ids = self . all_broker_ids ( ) <EOL> else : <EOL> broker_ids = self . broker_ids_for ( topic ) <EOL> broker_paths = map ( self . path_for_broker , broker_ids ) <EOL> broker_strings = map ( self . _zk_properties , broker_paths ) <EOL> broker_topic_paths = [ self . path_for_broker_topic ( broker_id , topic ) <EOL> for broker_id in broker_ids ] <EOL> num_parts = [ ] <EOL> for p in broker_topic_paths : <EOL> broker_parts = self . _zk_properties ( p ) if self . _zk . exists ( p ) else <NUM_LIT:1> <EOL> num_parts . append ( broker_parts ) <EOL> return list ( <EOL> chain . from_iterable ( <EOL> BrokerPartition . from_zk ( broker_id , broker_string , topic , n ) <EOL> for broker_id , broker_string , n <EOL> in zip ( broker_ids , broker_strings , num_parts ) <EOL> ) <EOL> ) <EOL> def offsets_state ( self , consumer_group ) : <EOL> """<STR_LIT>""" <EOL> def topic_offsets ( topic ) : <EOL> broker_parts_path = self . path_for_offsets ( consumer_group , topic ) <EOL> broker_parts = self . _zk_children ( broker_parts_path ) <EOL> return dict ( ( bp , int ( self . _zk_properties ( broker_parts_path + "<STR_LIT:/>" + bp ) ) ) <EOL> for bp in broker_parts ) <EOL> state = { } <EOL> topics = self . _zk_children ( "<STR_LIT>" . format ( consumer_group ) ) <EOL> for topic in topics : <EOL> state [ topic ] = topic_offsets ( topic ) <EOL> return state <EOL> def offsets_for ( self , consumer_group , consumer_id , broker_partitions ) : <EOL> """<STR_LIT>""" <EOL> UNKNOWN_OFFSET = <NUM_LIT:2> ** <NUM_LIT> - <NUM_LIT:1> <EOL> bps_to_offsets = { } <EOL> for bp in broker_partitions : <EOL> if self . _zk . exists ( self . path_for_topic ( bp . topic ) ) : <EOL> offset_path = self . path_for_offset ( consumer_group , <EOL> bp . topic , <EOL> bp . broker_id , <EOL> bp . partition ) <EOL> try : <EOL> offset = int ( self . _zk_properties ( offset_path ) ) <EOL> except zookeeper . NoNodeException as ex : <EOL> self . _create_path_if_needed ( offset_path , <NUM_LIT:0> ) <EOL> offset = <NUM_LIT:0> <EOL> bps_to_offsets [ bp ] = offset <EOL> return bps_to_offsets <EOL> def save_offsets_for ( self , consumer_group , bps_to_next_offsets ) : <EOL> bp_ids_to_offsets = sorted ( ( bp . id , offset ) <EOL> for bp , offset in bps_to_next_offsets . items ( ) ) <EOL> log . debug ( "<STR_LIT>" . format ( bp_ids_to_offsets ) ) <EOL> for bp , next_offset in sorted ( bps_to_next_offsets . items ( ) ) : <EOL> if self . _zk . exists ( self . path_for_topic ( bp . topic ) ) : <EOL> offset_path = self . path_for_offset ( consumer_group , <EOL> bp . topic , <EOL> bp . broker_id , <EOL> bp . partition ) <EOL> try : <EOL> offset_node = self . _zk . properties ( offset_path ) <EOL> except zookeeper . NoNodeException as ex : <EOL> self . _create_path_if_needed ( offset_path , bp ) <EOL> offset_node = self . _zk . properties ( offset_path ) <EOL> next_offset = <NUM_LIT:0> <EOL> if next_offset is not None : <EOL> log . debug ( "<STR_LIT>" % ( offset_node , next_offset ) ) <EOL> offset_node . set ( string_value = str ( next_offset ) ) <EOL> def broker_ids_for ( self , topic ) : <EOL> topic_path = self . path_for_topic ( topic ) <EOL> try : <EOL> topic_node_children = self . _zk_children ( topic_path ) <EOL> except zookeeper . NoNodeException : <EOL> log . warn ( u"<STR_LIT>" <EOL> . format ( topic_path , topic ) ) <EOL> return [ ] <EOL> return sorted ( int ( broker_id ) for broker_id in topic_node_children ) <EOL> def all_broker_ids ( self ) : <EOL> brokers_path = self . path_for_brokers ( ) <EOL> try : <EOL> topic_node_children = self . _zk_children ( brokers_path ) <EOL> except zookeeper . NoNodeException : <EOL> log . error ( u"<STR_LIT>" . format ( brokers_path ) ) <EOL> return [ ] <EOL> return sorted ( int ( broker_id ) for broker_id in topic_node_children ) <EOL> def consumer_ids_for ( self , topic , consumer_group ) : <EOL> """<STR_LIT>""" <EOL> cg_path = self . path_for_consumer_ids ( consumer_group ) <EOL> consumer_ids_in_group = sorted ( self . _zk_children ( cg_path ) ) <EOL> consumer_id_paths = [ self . path_for_consumer_id ( consumer_group , consumer_id ) <EOL> for consumer_id in consumer_ids_in_group ] <EOL> consumer_id_data = [ self . _zk_properties ( path ) <EOL> for path in consumer_id_paths ] <EOL> return [ consumer_id for consumer_id , data <EOL> in zip ( consumer_ids_in_group , consumer_id_data ) <EOL> if topic in data ] <EOL> def register_consumer ( self , consumer_group , consumer_id , topic ) : <EOL> """<STR_LIT>""" <EOL> self . _create_path_if_needed ( self . path_for_consumer_ids ( consumer_group ) , <EOL> consumer_id ) <EOL> consumer_id_path = self . path_for_consumer_id ( consumer_group , consumer_id ) <EOL> log . info ( "<STR_LIT>" <EOL> . format ( consumer_id , consumer_id_path ) ) <EOL> zookeeper . create ( self . _zk . handle , <EOL> consumer_id_path , <EOL> json . dumps ( { topic : <NUM_LIT:1> } ) , <EOL> ZKUtil . ACL , <EOL> zookeeper . EPHEMERAL ) <EOL> def _create_path_if_needed ( self , path , data = None ) : <EOL> """<STR_LIT>""" <EOL> if not path . startswith ( "<STR_LIT:/>" ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> def _build_path ( nodes ) : <EOL> return "<STR_LIT:/>" + "<STR_LIT:/>" . join ( nodes ) <EOL> nodes_to_create = path [ <NUM_LIT:1> : ] . split ( "<STR_LIT:/>" ) <EOL> created_so_far = [ ] <EOL> for node in nodes_to_create : <EOL> created_path = _build_path ( created_so_far ) <EOL> if node and node not in self . _zk . children ( created_path ) . data : <EOL> node_to_create = _build_path ( created_so_far + [ node ] ) <EOL> if isinstance ( data , basestring ) : <EOL> init_data = data <EOL> else : <EOL> init_data = json . dumps ( data ) <EOL> zookeeper . create ( self . _zk . handle , node_to_create , init_data , ZKUtil . ACL ) <EOL> created_so_far . append ( node ) <EOL> def path_for_broker_topic ( self , broker_id , topic_name ) : <EOL> return "<STR_LIT>" . format ( self . path_for_topic ( topic_name ) , broker_id ) <EOL> def path_for_brokers ( self ) : <EOL> return "<STR_LIT>" <EOL> def path_for_broker ( self , broker_id ) : <EOL> return "<STR_LIT>" . format ( broker_id ) <EOL> def path_for_topics ( self ) : <EOL> return "<STR_LIT>" <EOL> def path_for_topic ( self , topic ) : <EOL> return "<STR_LIT>" . format ( self . path_for_topics ( ) , topic ) <EOL> def path_for_offsets ( self , consumer_group , topic ) : <EOL> return ( "<STR_LIT>" . format ( consumer_group , topic ) ) <EOL> def path_for_offset ( self , consumer_group , topic , broker_id , partition ) : <EOL> path_for_offsets = self . path_for_offsets ( consumer_group , topic ) <EOL> return "<STR_LIT>" . format ( path_for_offsets , broker_id , partition ) <EOL> def path_for_consumer_ids ( self , consumer_group ) : <EOL> return u"<STR_LIT>" . format ( consumer_group ) <EOL> def path_for_consumer_id ( self , consumer_group , consumer_id ) : <EOL> return u"<STR_LIT>" . format ( self . path_for_consumer_ids ( consumer_group ) , <EOL> consumer_id ) <EOL> def _zk_properties ( self , path ) : <EOL> node_data = self . _zk . properties ( path ) . data <EOL> if '<STR_LIT>' in node_data : <EOL> return node_data [ '<STR_LIT>' ] <EOL> return node_data <EOL> def _zk_children ( self , path ) : <EOL> return self . _zk . children ( path ) . data <EOL> class ZKProducer ( object ) : <EOL> def __init__ ( self , zk_conn_str , topic , zk_timeout = None ) : <EOL> self . _id = uuid . uuid1 ( ) <EOL> self . _topic = topic <EOL> self . _bps_changed = False <EOL> self . _zk_util = ZKUtil ( zk_conn_str , zk_timeout ) <EOL> self . _brokers_watch = None <EOL> self . _topic_watch = None <EOL> self . _register_callbacks ( ) <EOL> self . detect_broker_partitions ( ) <EOL> broker_conn_info = frozenset ( ( bp . broker_id , bp . host , bp . port ) <EOL> for bp in self . _broker_partitions ) <EOL> self . _connections = dict ( ( broker_id , Kafka ( host , port ) ) <EOL> for broker_id , host , port in broker_conn_info ) <EOL> @ property <EOL> def topic ( self ) : <EOL> return self . _topic <EOL> @ property <EOL> def broker_partitions ( self ) : <EOL> return self . _broker_partitions [ : ] <EOL> def close ( self ) : <EOL> self . _zk_util . close ( ) <EOL> def send ( self , msgs ) : <EOL> if not msgs : <EOL> return <EOL> if not self . _all_callbacks_registered ( ) : <EOL> self . _register_callbacks ( ) <EOL> if self . _bps_changed : <EOL> self . detect_broker_partitions ( ) <EOL> broker_partition = random . choice ( self . _broker_partitions ) <EOL> kafka_conn = self . _connections [ broker_partition . broker_id ] <EOL> kafka_conn . produce ( self . topic , msgs , broker_partition . partition ) <EOL> bytes_sent = sum ( len ( m ) for m in msgs ) <EOL> log . debug ( self . _log_str ( u"<STR_LIT>" <EOL> . format ( bytes_sent , broker_partition ) ) ) <EOL> return broker_partition <EOL> def detect_broker_partitions ( self ) : <EOL> bps = self . _zk_util . broker_partitions_for ( self . topic , <EOL> force_partition_zero = True ) <EOL> if not bps : <EOL> raise NoAvailablePartitionsError ( u"<STR_LIT>" ) <EOL> self . _broker_partitions = bps <EOL> self . _bps_changed = False <EOL> def _unbalance ( self , nodes ) : <EOL> self . _bps_changed = True <EOL> def _register_callbacks ( self ) : <EOL> zk = self . _zk_util . _zk <EOL> path_for_brokers = self . _zk_util . path_for_brokers ( ) <EOL> path_for_topic = self . _zk_util . path_for_topic ( self . topic ) <EOL> if self . _brokers_watch is None and zk . exists ( path_for_brokers ) : <EOL> self . _brokers_watch = zk . children ( path_for_brokers ) ( self . _unbalance ) <EOL> if self . _topic_watch is None and zk . exists ( path_for_topic ) : <EOL> self . _topic_watch = zk . children ( path_for_topic ) ( self . _unbalance ) <EOL> log . debug ( "<STR_LIT>" <EOL> . format ( self . _id , sorted ( zk . watches . data . keys ( ) ) ) ) <EOL> def _all_callbacks_registered ( self ) : <EOL> """<STR_LIT>""" <EOL> return all ( [ self . _brokers_watch , self . _topic_watch ] ) <EOL> def _log_str ( self , s ) : <EOL> return u"<STR_LIT>" . format ( self . _id , s ) <EOL> def __del__ ( self ) : <EOL> self . close ( ) <EOL> class ZKConsumer ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , zk_conn , consumer_group , topic , autocommit = True , zk_timeout = None ) : <EOL> """<STR_LIT>""" <EOL> self . _id = self . _create_consumer_id ( consumer_group ) <EOL> self . _topic = topic <EOL> self . _consumer_group = consumer_group <EOL> self . _autocommit = autocommit <EOL> self . _zk_util = ZKUtil ( zk_conn , zk_timeout ) <EOL> self . _needs_rebalance = True <EOL> self . _broker_partitions = [ ] <EOL> self . _bps_to_next_offsets = { } <EOL> self . _rebalance_enabled = True <EOL> self . _topic_watch = None <EOL> self . _topics_watch = None <EOL> self . _consumers_watch = None <EOL> self . _brokers_watch = None <EOL> self . _register ( ) <EOL> self . rebalance ( ) <EOL> self . _stats = defaultdict ( lambda : ConsumerStats ( fetches = <NUM_LIT:0> , bytes = <NUM_LIT:0> , messages = <NUM_LIT:0> , max_fetch = <NUM_LIT:0> ) ) <EOL> @ property <EOL> def id ( self ) : return self . _id <EOL> @ property <EOL> def topic ( self ) : return self . _topic <EOL> @ property <EOL> def consumer_group ( self ) : return self . _consumer_group <EOL> @ property <EOL> def autocommit ( self ) : return self . _autocommit <EOL> @ property <EOL> def stats ( self ) : <EOL> '''<STR_LIT>''' <EOL> fetches = <NUM_LIT:0> <EOL> bytes = <NUM_LIT:0> <EOL> messages = <NUM_LIT:0> <EOL> max_fetch = <NUM_LIT:0> <EOL> for stats in self . _stats . values ( ) : <EOL> fetches += stats . fetches <EOL> bytes += stats . bytes <EOL> messages += stats . messages <EOL> max_fetch = max ( max_fetch , stats . max_fetch ) <EOL> return ConsumerStats ( fetches , bytes , messages , max_fetch ) <EOL> def stats_by_broker_partition ( self ) : <EOL> return dict ( self . _stats ) <EOL> @ property <EOL> def broker_partitions ( self ) : <EOL> if self . _needs_rebalance : <EOL> self . rebalance ( ) <EOL> return self . _broker_partitions <EOL> @ property <EOL> def brokers ( self ) : <EOL> return sorted ( frozenset ( bp . broker_id for bp in self . broker_partitions ) ) <EOL> def close ( self ) : <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> self . _zk_util . close ( ) <EOL> def simple_consumer ( self , bp_ids_to_offsets ) : <EOL> """<STR_LIT>""" <EOL> all_broker_partitions = self . _zk_util . broker_partitions_for ( self . topic ) <EOL> broker_partitions = dict ( ( bp , bp_ids_to_offsets ( bp . id ) ) <EOL> for bp in all_broker_partitions <EOL> if bp . id in bp_ids_to_offsets ) <EOL> return SimpleConsumer ( self . topic , broker_partitions ) <EOL> def fetch ( self , max_size = None , retry_limit = <NUM_LIT:3> , ignore_failures = False ) : <EOL> """<STR_LIT>""" <EOL> def needs_offset_values_from_zk ( bps_to_offsets ) : <EOL> """<STR_LIT>""" <EOL> return ( not bps_to_offsets ) or ( None in bps_to_offsets . values ( ) ) <EOL> log . debug ( "<STR_LIT>" . format ( self . id ) ) <EOL> if self . _needs_rebalance : <EOL> self . rebalance ( ) <EOL> bps_to_offsets = dict ( self . _bps_to_next_offsets ) <EOL> offsets_pulled_from_zk = False <EOL> if needs_offset_values_from_zk ( bps_to_offsets ) : <EOL> if bps_to_offsets : <EOL> bps_needing_offsets = [ bp for bp , offset in bps_to_offsets . items ( ) <EOL> if offset is None ] <EOL> else : <EOL> bps_needing_offsets = self . broker_partitions <EOL> bps_to_offsets . update ( self . _zk_util . offsets_for ( self . consumer_group , <EOL> self . _id , <EOL> bps_needing_offsets ) ) <EOL> offsets_pulled_from_zk = True <EOL> message_sets = [ ] <EOL> for bp in bps_to_offsets : <EOL> offset = bps_to_offsets [ bp ] <EOL> kafka = self . _connections [ bp . broker_id ] <EOL> partition = kafka . partition ( bp . topic , bp . partition ) <EOL> if offset is None : <EOL> offset = partition . latest_offset ( ) <EOL> try : <EOL> offsets_msgs = kafka . fetch ( bp . topic , <EOL> offset , <EOL> partition = bp . partition , <EOL> max_size = max_size ) <EOL> except OffsetOutOfRange as ex : <EOL> if offsets_pulled_from_zk : <EOL> log . error ( "<STR_LIT>" <EOL> . format ( offset , bp ) ) <EOL> offset = partition . latest_offset ( ) <EOL> log . error ( "<STR_LIT>" <EOL> . format ( offset , bp ) ) <EOL> offsets_msgs = kafka . fetch ( bp . topic , <EOL> offset , <EOL> partition = bp . partition , <EOL> max_size = max_size ) <EOL> else : <EOL> raise <EOL> except KafkaError as k_err : <EOL> if ignore_failures : <EOL> log . error ( "<STR_LIT>" . format ( bp ) ) <EOL> log . exception ( k_err ) <EOL> continue <EOL> else : <EOL> raise <EOL> msg_set = MessageSet ( bp , offset , offsets_msgs ) <EOL> old_stats = self . _stats [ bp ] <EOL> self . _stats [ bp ] = ConsumerStats ( fetches = old_stats . fetches + <NUM_LIT:1> , <EOL> bytes = old_stats . bytes + msg_set . size , <EOL> messages = old_stats . messages + len ( msg_set ) , <EOL> max_fetch = max ( old_stats . max_fetch , msg_set . size ) ) <EOL> message_sets . append ( msg_set ) <EOL> result = FetchResult ( sorted ( message_sets ) ) <EOL> for msg_set in result : <EOL> self . _bps_to_next_offsets [ msg_set . broker_partition ] = msg_set . next_offset <EOL> if self . _autocommit : <EOL> self . commit_offsets ( ) <EOL> return result <EOL> def commit_offsets ( self ) : <EOL> if self . _bps_to_next_offsets : <EOL> self . _zk_util . save_offsets_for ( self . consumer_group , <EOL> self . _bps_to_next_offsets ) <EOL> def poll ( self , <EOL> start_offsets = None , <EOL> end_offsets = None , <EOL> poll_interval = <NUM_LIT:1> , <EOL> max_size = None , <EOL> retry_limit = <NUM_LIT:3> ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> for msg_set in self . fetch ( max_size = max_size ) : <EOL> yield msg_set <EOL> time . sleep ( poll_interval ) <EOL> def _create_consumer_id ( self , consumer_group_id ) : <EOL> """<STR_LIT>""" <EOL> hostname = platform . node ( ) <EOL> ms_since_epoch = int ( time . time ( ) * <NUM_LIT:1000> ) <EOL> uuid_top_hex = uuid . uuid4 ( ) . hex [ : <NUM_LIT:8> ] <EOL> consumer_uuid = "<STR_LIT>" . format ( hostname , ms_since_epoch , uuid_top_hex ) <EOL> return "<STR_LIT>" . format ( consumer_group_id , consumer_uuid ) <EOL> def _register ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _zk_util . register_consumer ( self . consumer_group , self . id , self . topic ) <EOL> def rebalance ( self ) : <EOL> """<STR_LIT>""" <EOL> log . info ( ( "<STR_LIT>" + "<STR_LIT>" ) . format ( self . id , unicode ( self ) ) ) <EOL> if not self . _rebalance_enabled : <EOL> log . info ( "<STR_LIT>" ) <EOL> return <EOL> all_topic_consumers = self . _zk_util . consumer_ids_for ( self . topic , <EOL> self . consumer_group ) <EOL> all_broker_partitions = self . _zk_util . broker_partitions_for ( self . topic ) <EOL> try : <EOL> my_index = all_topic_consumers . index ( self . id ) <EOL> except ValueError : <EOL> msg_tmpl = "<STR_LIT>" + "<STR_LIT>" <EOL> raise ConsumerEntryNotFoundError ( <EOL> msg_tmpl . format ( self . id , self . topic , all_topic_consumers ) ) <EOL> bp_per_consumer = len ( all_broker_partitions ) / len ( all_topic_consumers ) <EOL> consumers_with_extra = range ( len ( all_broker_partitions ) % len ( all_topic_consumers ) ) <EOL> num_parts = bp_per_consumer + ( <NUM_LIT:1> if my_index in consumers_with_extra else <NUM_LIT:0> ) <EOL> if my_index == <NUM_LIT:0> : <EOL> start = <NUM_LIT:0> <EOL> elif my_index - <NUM_LIT:1> in consumers_with_extra : <EOL> start = my_index * ( bp_per_consumer + <NUM_LIT:1> ) <EOL> else : <EOL> start = ( len ( consumers_with_extra ) * ( bp_per_consumer + <NUM_LIT:1> ) ) + ( ( my_index - len ( consumers_with_extra ) ) * bp_per_consumer ) <EOL> self . _broker_partitions = all_broker_partitions [ start : start + num_parts ] <EOL> for bp in self . _bps_to_next_offsets . keys ( ) : <EOL> if bp not in self . _broker_partitions : <EOL> del self . _bps_to_next_offsets [ bp ] <EOL> for bp in self . _broker_partitions : <EOL> self . _bps_to_next_offsets . setdefault ( bp , None ) <EOL> broker_conn_info = frozenset ( ( bp . broker_id , bp . host , bp . port ) <EOL> for bp in self . _broker_partitions ) <EOL> self . _connections = dict ( ( broker_id , Kafka ( host , port ) ) <EOL> for broker_id , host , port in broker_conn_info ) <EOL> self . _register_callbacks ( ) <EOL> if self . _all_callbacks_registered ( ) : <EOL> self . _needs_rebalance = False <EOL> log . info ( "<STR_LIT>" . format ( self . id , unicode ( self ) ) ) <EOL> def disable_rebalance ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _rebalance_enabled = False <EOL> def enable_rebalance ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _rebalance_enabled = True <EOL> def _unbalance ( self , nodes ) : <EOL> """<STR_LIT>""" <EOL> self . _needs_rebalance = True <EOL> def _all_callbacks_registered ( self ) : <EOL> """<STR_LIT>""" <EOL> return all ( [ self . _consumers_watch , <EOL> self . _brokers_watch , <EOL> self . _topics_watch , <EOL> self . _topic_watch ] ) <EOL> def _register_callbacks ( self ) : <EOL> zk = self . _zk_util . _zk <EOL> path_for_consumers = self . _zk_util . path_for_consumer_ids ( self . consumer_group ) <EOL> path_for_brokers = self . _zk_util . path_for_brokers ( ) <EOL> path_for_topics = self . _zk_util . path_for_topics ( ) <EOL> path_for_topic = self . _zk_util . path_for_topic ( self . topic ) <EOL> if self . _consumers_watch is None and zk . exists ( path_for_consumers ) : <EOL> self . _consumers_watch = zk . children ( path_for_consumers ) ( self . _unbalance ) <EOL> if self . _brokers_watch is None and zk . exists ( path_for_brokers ) : <EOL> self . _brokers_watch = zk . children ( path_for_brokers ) ( self . _unbalance ) <EOL> if self . _topics_watch is None and zk . exists ( path_for_topics ) : <EOL> self . _topics_watch = zk . children ( path_for_topics ) ( self . _unbalance ) <EOL> if self . _topic_watch is None and zk . exists ( path_for_topic ) : <EOL> self . _topic_watch = zk . children ( path_for_topic ) ( self . _unbalance ) <EOL> log . debug ( "<STR_LIT>" <EOL> . format ( self . _id , sorted ( zk . watches . data . keys ( ) ) ) ) <EOL> def __unicode__ ( self ) : <EOL> bp_ids = [ bp . id for bp in self . _broker_partitions ] <EOL> return ( "<STR_LIT>" <EOL> . format ( self . id , bp_ids ) ) <EOL> def __del__ ( self ) : <EOL> self . close ( ) </s>
