docs: @SuppressWarnings("unchecked")
    private void handlePublish(PublishRequest request) {
      log.trace("Received {}", request);
  
      // If the request is for another session ID, this may be a session that was previously opened
      // for this client.
      if (request.session() != state.getSessionId().id()) {
        log.trace("Inconsistent session ID: {}", request.session());
        return;
      }
  
      // Store eventIndex in a local variable to prevent multiple volatile reads.
      long eventIndex = state.getEventIndex();
  
      // If the request event index has already been processed, return.
      if (request.eventIndex() <= eventIndex) {
        log.trace("Duplicate event index {}", request.eventIndex());
        return;
      }
  
      // If the request's previous event index doesn't equal the previous received event index,
      // respond with an undefined error and the last index received. This will cause the cluster
      // to resend events starting at eventIndex + 1.
      if (request.previousIndex() != eventIndex) {
        log.trace("Inconsistent event index: {}", request.previousIndex());
        ResetRequest resetRequest = ResetRequest.builder()
            .withSession(state.getSessionId().id())
            .withIndex(eventIndex)
            .build();
        log.trace("Sending {}", resetRequest);
        protocol.reset(memberSelector.members(), resetRequest);
        return;
      }
  
      // Store the event index. This will be used to verify that events are received in sequential order.
      state.setEventIndex(request.eventIndex());
  
      sequencer.sequenceEvent(request, () -> {
        for (PrimitiveEvent event : request.events()) {
          Set<Consumer<PrimitiveEvent>> listeners = eventListeners.get(event.type());
          if (listeners != null) {
            for (Consumer<PrimitiveEvent> listener : listeners) {
              listener.accept(event);
            }
          }
        }
      });
    }