#NEWLINE# def _message_generator( self ) : #NEWLINE# #INDENT# assert ( self . assignment ( ) or ( self . subscription ( ) is not None ) ) , 'No topic subscription or manual partition assignment' #NEWLINE# while ( time . time ( ) < self . _consumer_timeout ) : #NEWLINE# #INDENT# if self . _use_consumer_group ( ) : #NEWLINE# #INDENT# self . _coordinator . ensure_coordinator_known ( ) #NEWLINE# self . _coordinator . ensure_active_group ( ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( ( self . config [ 'group_id' ] is not None ) and ( self . config [ 'api_version' ] >= ( 0 , 8 , 2 ) ) ) : #NEWLINE# #INDENT# self . _coordinator . ensure_coordinator_known ( ) #NEWLINE# #UNINDENT# #UNINDENT# if ( not self . _subscription . has_all_fetch_positions ( ) ) : #NEWLINE# #INDENT# partitions = self . _subscription . missing_fetch_positions ( ) #NEWLINE# self . _update_fetch_positions ( partitions ) #NEWLINE# #UNINDENT# poll_ms = ( 1000 * ( self . _consumer_timeout - time . time ( ) ) ) #NEWLINE# if ( not self . _fetcher . in_flight_fetches ( ) ) : #NEWLINE# #INDENT# poll_ms = 0 #NEWLINE# #UNINDENT# self . _client . poll ( timeout_ms = poll_ms , sleep = True ) #NEWLINE# timeout_at = self . _next_timeout ( ) #NEWLINE# if ( self . _use_consumer_group ( ) and ( not self . assignment ( ) ) ) : #NEWLINE# #INDENT# sleep_time = max ( ( timeout_at - time . time ( ) ) , 0 ) #NEWLINE# if ( ( sleep_time > 0 ) and ( not self . _client . in_flight_request_count ( ) ) ) : #NEWLINE# #INDENT# log . debug ( 'No partitions assigned; sleeping for %s' , sleep_time ) #NEWLINE# time . sleep ( sleep_time ) #NEWLINE# continue #NEWLINE# #UNINDENT# #UNINDENT# if ( time . time ( ) > timeout_at ) : #NEWLINE# #INDENT# continue #NEWLINE# #UNINDENT# for msg in self . _fetcher : #NEWLINE# #INDENT# ( yield msg ) #NEWLINE# if ( time . time ( ) > timeout_at ) : #NEWLINE# #INDENT# log . debug ( 'internal iterator timeout - breaking for poll' ) #NEWLINE# break #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# self . _fetcher . init_fetches ( )