private void shutdown(){
  try {
    if (!runningFlag.get()) {
      while (!hold.isEmpty()) {
        PartitionReader p=hold.pollFirst();
        if (p == null) {
          break;
        }
 else {
          finish(p);
        }
      }
      while (!m_queue.isEmpty()) {
        PartitionReader p=m_queue.poll();
        if (p == null) {
          break;
        }
 else {
          finish(p);
        }
      }
      if (LOGGER.isInfoEnabled())       LOGGER.info(taskId + " is gracefully shutdown!");
    }
 else {
      LOGGER.error(taskId + " try to gracefully shutdown but runningFlag is true.");
    }
  }
 catch (  Throwable e) {
    LOGGER.error("Error occurs when gracefully shutdown kafkaConsumer thread " + taskId,e);
  }
}

private void shutdown ( ) { try { if ( ! runningFlag . get ( ) ) { while ( ! hold . isEmpty ( ) ) { PartitionReader p = hold . pollFirst ( ) ; if ( p == null ) { break ; } else { finish ( p ) ; } } while ( ! m_queue . isEmpty ( ) ) { PartitionReader p = m_queue . poll ( ) ; if ( p == null ) { break ; } else { finish ( p ) ; } } if ( LOGGER . isInfoEnabled ( ) ) LOGGER . info ( taskId + " is gracefully shutdown!" ) ; } else { LOGGER . error ( taskId + " try to gracefully shutdown but runningFlag is true." ) ; } } catch ( Throwable e ) { LOGGER . error ( "Error occurs when gracefully shutdown kafkaConsumer thread " + taskId , e ) ; } }
gracefully shutdown when kafka channel is unsubscribed commit and release all partitions
