question: 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 ) ; } } }
target:  BLOCKING Extend deadline for all messages which need it. CAUTION: If extensions can't keep up with wallclock then we'll never return.