#NEWLINE# def consume( topic , group , limit ) : #NEWLINE# #INDENT# if ( group is None ) : #NEWLINE# #INDENT# group = app . config [ 'DEFAULT_GROUP' ] . encode ( 'utf-8' ) #NEWLINE# #UNINDENT# consumer = get_kafka_consumer ( group , topic ) #NEWLINE# res = { #NEWLINE# #INDENT# 'group' : group . decode ( 'utf-8' ) , #NEWLINE# 'messages' : [ ] , #NEWLINE# #UNINDENT# } #NEWLINE# for ( n , message ) in enumerate ( consumer ) : #NEWLINE# #INDENT# res [ 'messages' ] . append ( message . message . value . decode ( 'utf-8' ) ) #NEWLINE# if ( n >= ( limit - 1 ) ) : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# #UNINDENT# consumer . commit ( ) #NEWLINE# consumer . stop ( ) #NEWLINE# return res