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;
}

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 ; }
log stats if time to do so .
