question: private void consume ( ) { consumer . subscribe ( topics ) ; while ( running . get ( ) ) { try { ConsumerRecords records = consumer . poll ( pollIntervalMs ) ; if ( records == null ) { continue ; } Iterator < ConsumerRecord < String , String > > iterator = records . iterator ( ) ; while ( iterator . hasNext ( ) ) { sendMessage ( iterator . next ( ) ) ; } } catch ( Exception ex ) { String error = "Error consuming messages from kafka" ; logger . error ( error , ex ) ; bus . send ( ConfigConstants . CONSUMER_ERROR_TOPIC , getErrorString ( error , ex . getMessage ( ) ) ) ; } } }
target:  Handles looping and consuming