protected void commitOffsets(OffsetStorageWriter offsetWriter,long commitTimeoutMs){
  long started=clock.currentTimeInMillis();
  long timeout=started + commitTimeoutMs;
  if (!offsetWriter.beginFlush())   return;
  Future<Void> flush=offsetWriter.doFlush(null);
  if (flush == null)   return;
  try {
    flush.get(Math.max(timeout - clock.currentTimeInMillis(),0),TimeUnit.MILLISECONDS);
    recordsSinceLastCommit=0;
    timeSinceLastCommitMillis=clock.currentTimeInMillis();
  }
 catch (  InterruptedException e) {
    logger.warn("Flush of {} offsets interrupted, cancelling",this);
    offsetWriter.cancelFlush();
  }
catch (  ExecutionException e) {
    logger.error("Flush of {} offsets threw an unexpected exception: ",this,e);
    offsetWriter.cancelFlush();
  }
catch (  TimeoutException e) {
    logger.error("Timed out waiting to flush {} offsets to storage",this);
    offsetWriter.cancelFlush();
  }
}

protected void commitOffsets ( OffsetStorageWriter offsetWriter , long commitTimeoutMs ) { long started = clock . currentTimeInMillis ( ) ; long timeout = started + commitTimeoutMs ; if ( ! offsetWriter . beginFlush ( ) ) return ; Future < Void > flush = offsetWriter . doFlush ( null ) ; if ( flush == null ) return ; try { flush . get ( Math . max ( timeout - clock . currentTimeInMillis ( ) , 0 ) , TimeUnit . MILLISECONDS ) ; recordsSinceLastCommit = 0 ; timeSinceLastCommitMillis = clock . currentTimeInMillis ( ) ; } catch ( InterruptedException e ) { logger . warn ( "Flush of {} offsets interrupted, cancelling" , this ) ; offsetWriter . cancelFlush ( ) ; } catch ( ExecutionException e ) { logger . error ( "Flush of {} offsets threw an unexpected exception: " , this , e ) ; offsetWriter . cancelFlush ( ) ; } catch ( TimeoutException e ) { logger . error ( "Timed out waiting to flush {} offsets to storage" , this ) ; offsetWriter . cancelFlush ( ) ; } }
flush offsets to storage .
