private void extend() throws IOException {
  while (true) {
    long nowMsSinceEpoch=now();
    List<String> assumeExpired=new ArrayList<>();
    List<String> toBeExtended=new ArrayList<>();
    List<String> toBeExpired=new ArrayList<>();
    for (    Map.Entry<String,InFlightState> entry : inFlight.entrySet()) {
      if (entry.getValue().ackDeadlineMsSinceEpoch - (ackTimeoutMs * ACK_SAFETY_PCT) / 100 > nowMsSinceEpoch) {
        break;
      }
      if (entry.getValue().ackDeadlineMsSinceEpoch - ACK_TOO_LATE.getMillis() < nowMsSinceEpoch) {
        assumeExpired.add(entry.getKey());
        continue;
      }
      if (entry.getValue().requestTimeMsSinceEpoch + PROCESSING_TIMEOUT.getMillis() < nowMsSinceEpoch) {
        toBeExpired.add(entry.getKey());
        continue;
      }
      toBeExtended.add(entry.getKey());
      if (toBeExtended.size() >= ACK_BATCH_SIZE) {
        break;
      }
    }
    if (assumeExpired.isEmpty() && toBeExtended.isEmpty() && toBeExpired.isEmpty()) {
      return;
    }
    if (!assumeExpired.isEmpty()) {
      numLateDeadlines.add(nowMsSinceEpoch,assumeExpired.size());
      for (      String ackId : assumeExpired) {
        inFlight.remove(ackId);
      }
    }
    if (!toBeExpired.isEmpty()) {
      numExpired.add(nowMsSinceEpoch,toBeExpired.size());
      for (      String ackId : toBeExpired) {
        inFlight.remove(ackId);
      }
    }
    if (!toBeExtended.isEmpty()) {
      long newDeadlineMsSinceEpoch=nowMsSinceEpoch + (ackTimeoutMs * ACK_EXTENSION_PCT) / 100;
      for (      String ackId : toBeExtended) {
        InFlightState state=inFlight.remove(ackId);
        inFlight.put(ackId,new InFlightState(state.requestTimeMsSinceEpoch,newDeadlineMsSinceEpoch));
      }
      extendBatch(nowMsSinceEpoch,toBeExtended);
    }
  }
}

private void extend ( ) throws IOException { while ( true ) { long nowMsSinceEpoch = now ( ) ; List < String > assumeExpired = new ArrayList < > ( ) ; List < String > toBeExtended = new ArrayList < > ( ) ; List < String > toBeExpired = new ArrayList < > ( ) ; for ( Map . Entry < String , InFlightState > entry : inFlight . entrySet ( ) ) { if ( entry . getValue ( ) . ackDeadlineMsSinceEpoch - ( ackTimeoutMs * ACK_SAFETY_PCT ) / 100 > nowMsSinceEpoch ) { break ; } if ( entry . getValue ( ) . ackDeadlineMsSinceEpoch - ACK_TOO_LATE . getMillis ( ) < nowMsSinceEpoch ) { assumeExpired . add ( entry . getKey ( ) ) ; continue ; } if ( entry . getValue ( ) . requestTimeMsSinceEpoch + PROCESSING_TIMEOUT . getMillis ( ) < nowMsSinceEpoch ) { toBeExpired . add ( entry . getKey ( ) ) ; continue ; } toBeExtended . add ( entry . getKey ( ) ) ; if ( toBeExtended . size ( ) >= ACK_BATCH_SIZE ) { break ; } } if ( assumeExpired . isEmpty ( ) && toBeExtended . isEmpty ( ) && toBeExpired . isEmpty ( ) ) { return ; } if ( ! assumeExpired . isEmpty ( ) ) { numLateDeadlines . add ( nowMsSinceEpoch , assumeExpired . size ( ) ) ; for ( String ackId : assumeExpired ) { inFlight . remove ( ackId ) ; } } if ( ! toBeExpired . isEmpty ( ) ) { numExpired . add ( nowMsSinceEpoch , toBeExpired . size ( ) ) ; for ( String ackId : toBeExpired ) { inFlight . remove ( ackId ) ; } } if ( ! toBeExtended . isEmpty ( ) ) { long newDeadlineMsSinceEpoch = nowMsSinceEpoch + ( ackTimeoutMs * ACK_EXTENSION_PCT ) / 100 ; for ( String ackId : toBeExtended ) { InFlightState state = inFlight . remove ( ackId ) ; inFlight . put ( ackId , new InFlightState ( state . requestTimeMsSinceEpoch , newDeadlineMsSinceEpoch ) ) ; } extendBatch ( nowMsSinceEpoch , toBeExtended ) ; } } }
blocking extend deadline for all messages which need it . caution : if extensions can ' t keep up with wallclock then we ' ll never return .
