code: private void completeSnapshot() {
      state.checkThread();
  
      // If a snapshot is pending to be persisted and the last completed index is greater than the
      // waiting snapshot index and no current or newer snapshot exists,
      // persist the snapshot and update the last snapshot index.
      if (pendingSnapshot != null && lastCompleted > pendingSnapshot.index()) {
        long snapshotIndex = pendingSnapshot.index();
        LOGGER.debug("{} - Completing snapshot {}", state.getCluster().member().address(), snapshotIndex);
        synchronized (pendingSnapshot) {
          Snapshot currentSnapshot = state.getSnapshotStore().currentSnapshot();
          if (currentSnapshot == null || snapshotIndex > currentSnapshot.index()) {
            pendingSnapshot.complete();
          } else {
            LOGGER.debug("{} - Discarding pending snapshot at index {} since the current snapshot is at index {}", state.getCluster().member().address(), pendingSnapshot.index(), currentSnapshot.index());
          }
          pendingSnapshot = null;
        }
  
        // Once the snapshot has been completed, snapshot dependent entries can be cleaned from the log.
        log.compactor().snapshotIndex(snapshotIndex);
        log.compactor().compact();
      }
    }
docs: Completes a snapshot of the state machine state.
  <p>
  When a snapshot of the state machine is taken, the snapshot isn't immediately made available for
  recovery or replication. Session events are dependent on original commands being retained in the log
  for fault tolerance. Thus, a snapshot cannot be completed until all prior events have been received
  by clients. So, we take a snapshot of the state machine state and complete the snapshot only after
  prior events have been received.