#NEWLINE# def __init__( self , * topics , **configs ) : #NEWLINE# #INDENT# self . config = copy . copy ( self . DEFAULT_CONFIG ) #NEWLINE# for key in self . config : #NEWLINE# #INDENT# if ( key in configs ) : #NEWLINE# #INDENT# self . config [ key ] = configs . pop ( key ) #NEWLINE# #UNINDENT# #UNINDENT# assert ( not configs ) , ( 'Unrecognized configs: %s' % configs ) #NEWLINE# deprecated = { #NEWLINE# #INDENT# 'smallest' : 'earliest' , #NEWLINE# 'largest' : 'latest' , #NEWLINE# #UNINDENT# } #NEWLINE# if ( self . config [ 'auto_offset_reset' ] in deprecated ) : #NEWLINE# #INDENT# new_config = deprecated [ self . config [ 'auto_offset_reset' ] ] #NEWLINE# log . warning ( 'use auto_offset_reset=%s (%s is deprecated)' , new_config , self . config [ 'auto_offset_reset' ] ) #NEWLINE# self . config [ 'auto_offset_reset' ] = new_config #NEWLINE# #UNINDENT# metrics_tags = { #NEWLINE# #INDENT# 'client-id' : self . config [ 'client_id' ] , #NEWLINE# #UNINDENT# } #NEWLINE# metric_config = MetricConfig ( samples = self . config [ 'metrics_num_samples' ] , time_window_ms = self . config [ 'metrics_sample_window_ms' ] , tags = metrics_tags ) #NEWLINE# reporters = [ reporter ( ) for reporter in self . config [ 'metric_reporters' ] ] #NEWLINE# reporters . append ( DictReporter ( 'kafka.consumer' ) ) #NEWLINE# self . _metrics = Metrics ( metric_config , reporters ) #NEWLINE# metric_group_prefix = 'consumer' #NEWLINE# self . _client = KafkaClient ( ** self . config ) #NEWLINE# if ( self . config [ 'api_version' ] == 'auto' ) : #NEWLINE# #INDENT# self . config [ 'api_version' ] = self . _client . check_version ( ) #NEWLINE# #UNINDENT# assert ( self . config [ 'api_version' ] in ( '0.9' , '0.8.2' , '0.8.1' , '0.8.0' ) ) , 'Unrecognized api version' #NEWLINE# self . config [ 'api_version' ] = tuple ( map ( int , self . config [ 'api_version' ] . split ( '.' ) ) ) #NEWLINE# self . _subscription = SubscriptionState ( self . config [ 'auto_offset_reset' ] ) #NEWLINE# self . _fetcher = Fetcher ( self . _client , self . _subscription , new_config . _metrics , metric_group_prefix , ** self . config ) #NEWLINE# self . _coordinator = ConsumerCoordinator ( self . _client , self . _subscription , self . _metrics , metric_group_prefix , assignors = self . config [ 'partition_assignment_strategy' ] , ** self . config ) #NEWLINE# self . _closed = False #NEWLINE# self . _iterator = None #NEWLINE# self . _consumer_timeout = float ( 'inf' ) #NEWLINE# if topics : #NEWLINE# #INDENT# self . _subscription . subscribe ( topics = topics ) #NEWLINE# self . _client . set_topics ( topics )