public void nackAll(PubsubReader<T> reader) throws IOException {
  checkState(this.reader == null,"Cannot nackAll on persisting checkpoint");
  List<String> batchYetToAckIds=new ArrayList<>(Math.min(notYetReadIds.size(),ACK_BATCH_SIZE));
  for (  String ackId : notYetReadIds) {
    batchYetToAckIds.add(ackId);
    if (batchYetToAckIds.size() >= ACK_BATCH_SIZE) {
      long nowMsSinceEpoch=now(reader);
      reader.nackBatch(nowMsSinceEpoch,batchYetToAckIds);
      batchYetToAckIds.clear();
    }
  }
  if (!batchYetToAckIds.isEmpty()) {
    long nowMsSinceEpoch=now(reader);
    reader.nackBatch(nowMsSinceEpoch,batchYetToAckIds);
  }
}

public void nackAll ( PubsubReader < T > reader ) throws IOException { checkState ( this . reader == null , "Cannot nackAll on persisting checkpoint" ) ; List < String > batchYetToAckIds = new ArrayList < > ( Math . min ( notYetReadIds . size ( ) , ACK_BATCH_SIZE ) ) ; for ( String ackId : notYetReadIds ) { batchYetToAckIds . add ( ackId ) ; if ( batchYetToAckIds . size ( ) >= ACK_BATCH_SIZE ) { long nowMsSinceEpoch = now ( reader ) ; reader . nackBatch ( nowMsSinceEpoch , batchYetToAckIds ) ; batchYetToAckIds . clear ( ) ; } } if ( ! batchYetToAckIds . isEmpty ( ) ) { long nowMsSinceEpoch = now ( reader ) ; reader . nackBatch ( nowMsSinceEpoch , batchYetToAckIds ) ; } }
blocking nack all messages which have been read from pubsub but not passed downstream . this way pubsub will send them again promptly .
