question: private void stats ( ) { long nowMsSinceEpoch = now ( ) ; if ( lastLogTimestampMsSinceEpoch < 0 ) { lastLogTimestampMsSinceEpoch = nowMsSinceEpoch ; return ; } long deltaMs = nowMsSinceEpoch - lastLogTimestampMsSinceEpoch ; if ( deltaMs < LOG_PERIOD . getMillis ( ) ) { return ; } String messageSkew = "unknown" ; long minTimestamp = minReceivedTimestampMsSinceEpoch . get ( nowMsSinceEpoch ) ; long maxTimestamp = maxReceivedTimestampMsSinceEpoch . get ( nowMsSinceEpoch ) ; if ( minTimestamp < Long . MAX_VALUE && maxTimestamp > Long . MIN_VALUE ) { messageSkew = ( maxTimestamp - minTimestamp ) + "ms" ; } String watermarkSkew = "unknown" ; long minWatermark = minWatermarkMsSinceEpoch . get ( nowMsSinceEpoch ) ; long maxWatermark = maxWatermarkMsSinceEpoch . get ( nowMsSinceEpoch ) ; if ( minWatermark < Long . MAX_VALUE && maxWatermark > Long . MIN_VALUE ) { watermarkSkew = ( maxWatermark - minWatermark ) + "ms" ; } String oldestInFlight = "no" ; String oldestAckId = Iterables . getFirst ( inFlight . keySet ( ) , null ) ; if ( oldestAckId != null ) { oldestInFlight = ( nowMsSinceEpoch - inFlight . get ( oldestAckId ) . requestTimeMsSinceEpoch ) + "ms" ; } LOG . info ( "Pubsub {} has " + "{} received messages, " + "{} current unread messages, " + "{} current unread bytes, " + "{} current in-flight msgs, " + "{} oldest in-flight, " + "{} current in-flight checkpoints, " + "{} max in-flight checkpoints, " + "{}B/s recent read, " + "{} recent received, " + "{} recent extended, " + "{} recent late extended, " + "{} recent ACKed, " + "{} recent NACKed, " + "{} recent expired, " + "{} recent message timestamp skew, " + "{} recent watermark skew, " + "{} recent late messages, " + "{} last reported watermark" , outer . outer . subscription , numReceived , notYetRead . size ( ) , notYetReadBytes , inFlight . size ( ) , oldestInFlight , numInFlightCheckpoints . get ( ) , maxInFlightCheckpoints , numReadBytes . get ( nowMsSinceEpoch ) / ( SAMPLE_PERIOD . getMillis ( ) / 1000L ) , numReceivedRecently . get ( nowMsSinceEpoch ) , numExtendedDeadlines . get ( nowMsSinceEpoch ) , numLateDeadlines . get ( nowMsSinceEpoch ) , numAcked . get ( nowMsSinceEpoch ) , numNacked . get ( nowMsSinceEpoch ) , numExpired . get ( nowMsSinceEpoch ) , messageSkew , watermarkSkew , numLateMessages . get ( nowMsSinceEpoch ) , new Instant ( lastWatermarkMsSinceEpoch ) ) ; lastLogTimestampMsSinceEpoch = nowMsSinceEpoch ; }
target:  Log stats if time to do so.