code: @Override
  	protected final void acknowledgeIDs(long checkpointId, Set<UId> uniqueIds) {
  		LOG.debug("Acknowledging ids for checkpoint {}", checkpointId);
  		Iterator<Tuple2<Long, List<SessionId>>> iterator = sessionIdsPerSnapshot.iterator();
  		while (iterator.hasNext()) {
  			final Tuple2<Long, List<SessionId>> next = iterator.next();
  			long id = next.f0;
  			if (id <= checkpointId) {
  				acknowledgeSessionIDs(next.f1);
  				// remove ids for this session
  				iterator.remove();
  			}
  		}
  	}
docs: Acknowledges the session ids.
  @param checkpointId The id of the current checkout to acknowledge ids for.
  @param uniqueIds The checkpointed unique ids which are ignored here. They only serve as a
  means of de-duplicating messages when the acknowledgment after a checkpoint
  fails.