code: private void scheduleAcknowledgment(final String topic) {
      if (!acknowledgeScheduled.has(topic)) {
        acknowledgeScheduled.set(topic, true);
        Platform.scheduler().scheduleDelay(acknowledgeDelayMillis, new Handler<Void>() {
          @Override
          public void handle(Void event) {
            if (acknowledgeScheduled.has(topic)) {
              acknowledgeScheduled.remove(topic);
              // Check we're still out of date, and not already catching up.
              double knownHeadSequence = knownHeadSequences.getNumber(topic);
              double currentSequence = currentSequences.getNumber(topic);
              if (knownHeadSequence > currentSequence
                  && (!acknowledgeNumbers.has(topic) || knownHeadSequence > acknowledgeNumbers
                      .getNumber(topic))) {
                acknowledgeNumbers.set(topic, knownHeadSequence);
                log.log(Level.CONFIG, "Catching up to " + knownHeadSequence);
                catchup(topic, currentSequence);
              } else {
                log.log(Level.FINE, "No need to catchup");
              }
            }
          }
        });
      }
    }
docs: Acknowledgment Number is the next sequence number that the receiver is expecting