code: @Synchronized
      private void checkpoint(PositionCounter pc) {
          running.remove(pc);
          completed.add(pc);
  
          final PositionCounter smallest = running.isEmpty() ? MAX : running.first();
          final List<PositionCounter> checkpointCandidates = completed.stream()
                  .filter(x -> positionCounterComparator.compare(x, smallest) < 0).collect(Collectors.toList());
          if (checkpointCandidates.size() > 0) {
              final PositionCounter checkpointPosition = checkpointCandidates.get(checkpointCandidates.size() - 1);
              completed.removeAll(checkpointCandidates);
              checkpoint.set(checkpointPosition);
          }
      }
docs: This method maintains a sorted list of position for requests currently being processed.
  As processing of each request completes, it is removed from the sorted list and moved to
  completed list.
  Completed is also a sorted list that contains all completed requests greater than smallest running position.
  In other words, we maintain all requests from smallest processing to current position in either running or completed
  sorted list.
  Note: Smallest position will always be in the running list.
  We also maintain a single checkpoint, which is the highest completed position smaller than smallest running position.
  
  @param pc position for which processing completed