code: void onSeqnoPersisted(final short vbucket, final long seqno) {
      final Queue<BufferedEvent> queue = partitionQueues.get(vbucket);
  
      synchronized (queue) {
        // while the head of the queue has a seqno <= the given seqno
        while (!queue.isEmpty() && Long.compareUnsigned(queue.peek().seqno, seqno) < 1) {
          final BufferedEvent event = queue.poll();
  
          try {
            switch (event.type) {
              case DATA: // mutation, deletion, expiration
                dataEventHandler.onEvent(event.flowController, event.event);
                break;
  
              case CONTROL: // snapshot
                controlEventHandler.onEvent(event.flowController, event.event);
                break;
  
              case STREAM_END_OK:
                eventBus.publish(new StreamEndEvent(vbucket, StreamEndReason.OK));
                break;
  
              default:
                throw new RuntimeException("Unexpected event type: " + event.type);
            }
  
          } catch (Throwable t) {
            LOGGER.error("Event handler threw exception", t);
          }
        }
      }
    }
docs: Send to the wrapped handler all events with sequence numbers <= the given sequence number.