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()));
    }
  }
}

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 ( ) ) ) ; } } }
handles looping and consuming
