private void readRequest() throws IOException {
  zkServer.processPacket(this, incomingBuffer);
}

if (h.getType() == OpCode.auth) {
  LOG.info("got auth packet " + cnxn.getRemoteSocketAddress());
  // Create authentication packet
  AuthPacket authPacket = new AuthPacket();
  // Convert byte buffer data to authentication packet object
  ByteBufferInputStream.byteBuffer2Record(incomingBuffer, authPacket);
  // Get authentication scheme
  String scheme = authPacket.getScheme();
  // Get authentication provider based on the scheme
  AuthenticationProvider ap = ProviderRegistry.getProvider(scheme);
  // Set initial authentication status code to AUTHFAILED
  Code authReturn = KeeperException.Code.AUTHFAILED;
  if (ap != null) {
    try {
      // Get status code from authentication provider
      authReturn = ap.handleAuthentication(cnxn, authPacket.getAuth());
    } catch (RuntimeException e) {
      LOG.warn("Caught runtime exception from AuthenticationProvider: " + scheme
               + " due to " + e);
      // Set authentication status code to AUTHFAILED
      authReturn = KeeperException.Code.AUTHFAILED;
    }
  }
  // If authentication status code is OK, send response
  if (authReturn == KeeperException.Code.OK) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Authentication succeeded for scheme: " + scheme);
    }
    LOG.info("auth success " + cnxn.getRemoteSocketAddress());
    ReplyHeader rh = new ReplyHeader(h.getXid(), 0, KeeperException.Code.OK.intValue());
    cnxn.sendResponse(rh, null, null);
  }
  // If authentication status code is not OK
  else {
    if (ap == null) {
      LOG.warn("No authentication provider for scheme: " + scheme + " has "
               + ProviderRegistry.listProviders());
    } else {
      LOG.warn("Authentication failed for scheme: " + scheme);
    }
    // Assemble and send response information
    ReplyHeader rh =
      new ReplyHeader(h.getXid(), 0, KeeperException.Code.AUTHFAILED.intValue());
    cnxn.sendResponse(rh, null, null);
    // Send close connection
    cnxn.sendBuffer(ServerCnxnFactory.closeConn);
    // Stop receiving data
    cnxn.disableRecv();
  }
  // End processing
  return;
}

private void processSasl(ByteBuffer incomingBuffer, ServerCnxn cnxn,
                         RequestHeader requestHeader) throws IOException {
  LOG.debug("Responding to client SASL token.");
  // Create SASL request object
  GetSASLRequest clientTokenRecord = new GetSASLRequest();
  // Convert byte buffer data to SASL request object
  ByteBufferInputStream.byteBuffer2Record(incomingBuffer, clientTokenRecord);
  // Get client token from SASL request object
  byte[] clientToken = clientTokenRecord.getToken();
  LOG.debug("Size of client SASL token: " + clientToken.length);
  // Response token
  byte[] responseToken = null;
  try {
    // Get SASL server from service connection object
    ZooKeeperSaslServer saslServer = cnxn.zooKeeperSaslServer;
    try {
      // Create response token through SASL server
      responseToken = saslServer.evaluateResponse(clientToken);
      // Check if SASL server processing is complete
      if (saslServer.isComplete()) {
        // Get authorization ID from SASL server
        String authorizationID = saslServer.getAuthorizationID();
        LOG.info("adding SASL authorization for authorizationID: " + authorizationID);
        // Add security information to service connection object
        cnxn.addAuthInfo(new Id("sasl", authorizationID));
        // If zookeeper.superUser property is not empty and authorization ID matches, add super user security information
        if (System.getProperty("zookeeper.superUser") != null &&
            authorizationID.equals(System.getProperty("zookeeper.superUser"))) {
          cnxn.addAuthInfo(new Id("super", ""));
        }
      }
    }
    // Exception handling
    catch (SaslException e) {
      LOG.warn("Client {} failed to SASL authenticate: {}",
               cnxn.getRemoteSocketAddress(), e);
      // Allow SASL authentication failure, client doesn't require SASL verification, output warning log
      if (shouldAllowSaslFailedClientsConnect() && !shouldRequireClientSaslAuth()) {
        LOG.warn("Maintaining client connection despite SASL authentication failure.");
      } else {
        // Confirm error value
        int error;
        // SASL verification required
        if (shouldRequireClientSaslAuth()) {
          LOG.warn(
            "Closing client connection due to server requires client SASL authentication,"
            +
            "but client SASL authentication has failed, or client is not configured with SASL "
            +
            "authentication.");
          // Session closed, SASL authentication failed
          error = Code.SESSIONCLOSEDREQUIRESASLAUTH.intValue();
        } else {
          LOG.warn("Closing client connection due to SASL authentication failure.");
          // Authentication failed
          error = Code.AUTHFAILED.intValue();
        }

        // Assemble response
        ReplyHeader replyHeader = new ReplyHeader(requestHeader.getXid(), 0, error);
        // Send response
        cnxn.sendResponse(replyHeader, new SetSASLResponse(null), "response");
        // Send close session message
        cnxn.sendCloseSession();
        // Stop receiving data
        cnxn.disableRecv();
        return;
      }
    }
  } catch (NullPointerException e) {
    LOG.error(
      "cnxn.saslServer is null: cnxn object did not initialize its saslServer properly.");
  }
  if (responseToken != null) {
    LOG.debug("Size of server SASL response: " + responseToken.length);
  }

  // SASL verification passed, send success message
  ReplyHeader replyHeader = new ReplyHeader(requestHeader.getXid(), 0, Code.OK.intValue());
  Record record = new SetSASLResponse(responseToken);
  cnxn.sendResponse(replyHeader, record, "response");
}

else {
  // SASL authentication is required and has not passed SASL authentication
  if (shouldRequireClientSaslAuth() && !hasCnxSASLAuthenticated(cnxn)) {
    // Assemble response information and send it
    ReplyHeader replyHeader = new ReplyHeader(h.getXid(), 0,
                                              Code.SESSIONCLOSEDREQUIRESASLAUTH.intValue());
    cnxn.sendResponse(replyHeader, null, "response");
    // Send session close information
    cnxn.sendCloseSession();
    // No longer receive data
    cnxn.disableRecv();
  }
  // SASL authentication is not required or SASL authentication has passed
  else {
    Request si = new Request(cnxn, cnxn.getSessionId(), h.getXid(), h.getType(),
                             incomingBuffer, cnxn.getAuthInfo());
    si.setOwner(ServerCnxn.me);
    // Confirm if it's a local session, if so, set isLocalSession to true
    setLocalSessionFlag(si);
    // Submit request for processing
    submitRequest(si);
  }
}

public void submitRequest(Request si) {
  // If the first request processor is null
  if (firstProcessor == null) {
    synchronized (this) {
      try {
        // Since all requests are passed to the request processor, it should wait for the request processor chain to be set up. The state will be updated to RUNNING after setup.
        while (state == State.INITIAL) {
          wait(1000);
        }
      } catch (InterruptedException e) {
        LOG.warn("Unexpected interruption", e);
      }
      if (firstProcessor == null || state != State.RUNNING) {
        throw new RuntimeException("Not started");
      }
    }
  }
  try {
    // Try to process session-related information
    touch(si.cnxn);
    // Validate request type
    boolean validpacket = Request.isValid(si.type);
    // Request type validation passed
    if (validpacket) {
      // Process request
      firstProcessor.processRequest(si);
      // If connection object exists
      if (si.cnxn != null) {
        // Increment the number of requests being processed
        incInProcess();
      }
    }
    // Request type validation failed
    else {
      LOG.warn("Received packet at server of unknown type " + si.type);
      // Request processor for sending error code response
      new UnimplementedRequestProcessor().processRequest(si);
    }
  } catch (MissingSessionException e) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Dropping request: " + e.getMessage());
    }
  } catch (RequestProcessorException e) {
    LOG.error("Unable to process request:" + e.getMessage(), e);
  }
}

protected void setupRequestProcessors() {
  RequestProcessor finalProcessor = new FinalRequestProcessor(this);
  RequestProcessor syncProcessor = new SyncRequestProcessor(this,
                                                            finalProcessor);
  ((SyncRequestProcessor) syncProcessor).start();
  firstProcessor = new PrepRequestProcessor(this, syncProcessor);
  ((PrepRequestProcessor) firstProcessor).start();
}

public void processRequest(Request request) {
    submittedRequests.add(request);
}

LinkedBlockingQueue<Request> submittedRequests = new LinkedBlockingQueue<Request>();

@Override
public void run() {
  try {
    // Infinite loop
    while (true) {
      // Get a request from the member variable submittedRequests
      Request request = submittedRequests.take();
      // Set the trace type to CLIENT_REQUEST_TRACE_MASK
      long traceMask = ZooTrace.CLIENT_REQUEST_TRACE_MASK;
      // If the request type is ping, set the trace type to CLIENT_PING_TRACE_MASK
      if (request.type == OpCode.ping) {
        traceMask = ZooTrace.CLIENT_PING_TRACE_MASK;
      }
      if (LOG.isTraceEnabled()) {
        ZooTrace.logRequest(LOG, traceMask, 'P', request, "");
      }
      // If the current request is the same as the death request, end the infinite loop and stop working
      if (Request.requestOfDeath == request) {
        break;
      }
      // Process the request
      pRequest(request);
    }
  } catch (RequestProcessorException e) {
    if (e.getCause() instanceof XidRolloverException) {
      LOG.info(e.getCause().getMessage());
    }
    // Handle exception
    handleException(this.getName(), e);
  } catch (Exception e) {
    handleException(this.getName(), e);
  }
  LOG.info("PrepRequestProcessor exited loop!");
}

private List<ACL> fixupACL(String path, List<Id> authInfo, List<ACL> acls)
  throws KeeperException.InvalidACLException {
  // Filter out duplicate data from the acls parameter
  List<ACL> uniqacls = removeDuplicates(acls);
  // Container for storing return values
  LinkedList<ACL> rv = new LinkedList<ACL>();
  // Throw an exception if the filtered ACL count is empty
  if (uniqacls == null || uniqacls.size() == 0) {
    throw new KeeperException.InvalidACLException(path);
  }
  // Iterate through the filtered acl collection
  for (ACL a : uniqacls) {
    LOG.debug("Processing ACL: {}", a);
    // Throw an exception if the current element is null
    if (a == null) {
      throw new KeeperException.InvalidACLException(path);
    }
    // Extract id
    Id id = a.getId();
    // Throw an exception if id is null or the scheme stored in id is null
    if (id == null || id.getScheme() == null) {
      throw new KeeperException.InvalidACLException(path);
    }
    // Add to the result set if the scheme stored in id is "world" and the id variable in id is "anyone"
    if (id.getScheme().equals("world") && id.getId().equals("anyone")) {
      rv.add(a);
    }
    // Perform authentication logic if the scheme stored in id is "auth"
    else if (id.getScheme().equals("auth")) {
      // Authentication pass flag
      boolean authIdValid = false;
      for (Id cid : authInfo) {
        // Get security authenticator based on security scheme
        AuthenticationProvider ap =
          ProviderRegistry.getProvider(cid.getScheme());

        if (ap == null) {
          LOG.error("Missing AuthenticationProvider for "
                    + cid.getScheme());
        }
        // Authentication passed
        else if (ap.isAuthenticated()) {
          authIdValid = true;
          rv.add(new ACL(a.getPerms(), cid));
        }
      }
      // Throw an exception if the authentication pass flag is false
      if (!authIdValid) {
        throw new KeeperException.InvalidACLException(path);
      }
    }
    else {
      // Get security authenticator based on security scheme
      AuthenticationProvider ap = ProviderRegistry.getProvider(id.getScheme());
      // Throw an exception if the security authenticator is null or id verification fails
      if (ap == null || !ap.isValid(id.getId())) {
        throw new KeeperException.InvalidACLException(path);
      }
      rv.add(a);
    }
  }
  return rv;
}

private ChangeRecord getRecordForPath(String path) throws KeeperException.NoNodeException {
  ChangeRecord lastChange = null;
  // Lock the outstanding changes
  synchronized (zks.outstandingChanges) {
    // Get change record from outstandingChangesForPath object
    lastChange = zks.outstandingChangesForPath.get(path);
    if (lastChange == null) {
      // Get the data node corresponding to the path from the zk database
      DataNode n = zks.getZKDatabase().getNode(path);
      // If the data node is not null
      if (n != null) {
        Set<String> children;
        synchronized (n) {
          // Get the child nodes of the current node
          children = n.getChildren();
        }
        // Rewrite change record
        lastChange = new ChangeRecord(-1, path, n.stat, children.size(),
                                      zks.getZKDatabase().aclForNode(n));
      }
    }
  }
  if (lastChange == null || lastChange.stat == null) {
    throw new KeeperException.NoNodeException(path);
  }
  // Return
  return lastChange;
}

final Deque<ChangeRecord> outstandingChanges = new ArrayDeque<>();
final HashMap<String, ChangeRecord> outstandingChangesForPath = new HashMap<String, ChangeRecord>();

private void addChangeRecord(ChangeRecord c) {
  synchronized (zks.outstandingChanges) {
    zks.outstandingChanges.add(c);
    zks.outstandingChangesForPath.put(c.path, c);
  }
}

static void checkACL(ZooKeeperServer zks, List<ACL> acl, int perm,
                     List<Id> ids) throws KeeperException.NoAuthException {
  // Skip ACL authentication
  if (skipACL) {
    return;
  }
  if (LOG.isDebugEnabled()) {
    LOG.debug("Permission requested: {} ", perm);
    LOG.debug("ACLs for node: {}", acl);
    LOG.debug("Client credentials: {}", ids);
  }
  // ACL list is empty or ACL count is zero
  if (acl == null || acl.size() == 0) {
    return;
  }
  // Process id collection
  for (Id authId : ids) {
    // If one of the elements in the id collection has a super scheme, return
    if (authId.getScheme().equals("super")) {
      return;
    }
  }
  // Process ACL collection
  for (ACL a : acl) {
    Id id = a.getId();
    if ((a.getPerms() & perm) != 0) {
      // Allow if scheme is world and id data is anyone
      if (id.getScheme().equals("world")
          && id.getId().equals("anyone")) {
        return;
      }
      // Get security authenticator based on security scheme
      AuthenticationProvider ap = ProviderRegistry.getProvider(id
                                                               .getScheme());
      // Security authenticator is not null
      if (ap != null) {
        // Process parameter ids
        for (Id authId : ids) {
          // Check if the scheme in the current id is the same as the scheme in the ACL id
          // Security authenticator matches the current id and the id in the ACL
          if (authId.getScheme().equals(id.getScheme())
              && ap.matches(authId.getId(), id.getId())) {
            return;
          }
        }
      }
    }
  }
  // Throw exception
  throw new KeeperException.NoAuthException();
}

private void validatePath(String path, long sessionId) throws BadArgumentsException {
  try {
    PathUtils.validatePath(path);
  } catch (IllegalArgumentException ie) {
    LOG.info("Invalid path {} with session 0x{}, reason: {}",
             path, Long.toHexString(sessionId), ie.getMessage());
    throw new BadArgumentsException(path);
  }
}

c > '\u0000' && c <= '\u001f'
        || c >= '\u007f' && c <= '\u009F'
        || c >= '\ud800' && c <= '\uf8ff'
        || c >= '\ufff0' && c <= '\uffff'

private void pRequest2TxnCreate(int type, Request request, Record record, boolean deserialize)
  throws IOException, KeeperException {
  // Check if deserialization is needed, if so, convert the request to record
  if (deserialize) {
    ByteBufferInputStream.byteBuffer2Record(request.request, record);
  }

  // Flag indicating the creation type
  int flags;
  // Node path
  String path;
  // ACL information
  List<ACL> acl;
  // Data information
  byte[] data;
  // Expiration time
  long ttl;
  // Type is creating a node with expiration time
  if (type == OpCode.createTTL) {
    CreateTTLRequest createTtlRequest = (CreateTTLRequest) record;
    flags = createTtlRequest.getFlags();
    path = createTtlRequest.getPath();
    acl = createTtlRequest.getAcl();
    data = createTtlRequest.getData();
    ttl = createTtlRequest.getTtl();
  }
  else {
    CreateRequest createRequest = (CreateRequest) record;
    flags = createRequest.getFlags();
    path = createRequest.getPath();
    acl = createRequest.getAcl();
    data = createRequest.getData();
    ttl = -1;
  }
  // Convert flags to creation type
  CreateMode createMode = CreateMode.fromFlag(flags);
  // Validate if the create request is legal
  validateCreateRequest(path, createMode, request, ttl);
  // Validate the creation path
  String parentPath = validatePathForCreate(path, request.sessionId);

  // Process ACL
  List<ACL> listACL = fixupACL(path, request.authInfo, acl);
  // Get the data record of the parent node
  ChangeRecord parentRecord = getRecordForPath(parentPath);

  // Validate ACL
  checkACL(zks, parentRecord.acl, ZooDefs.Perms.CREATE, request.authInfo);
  // Get the cversion (creation version number) of the parent node
  int parentCVersion = parentRecord.stat.getCversion();
  // Check if it's a sequential creation
  if (createMode.isSequential()) {
    path = path + String.format(Locale.ENGLISH, "%010d", parentCVersion);
  }
  // Validate if the path is legal
  validatePath(path, request.sessionId);
  try {
    // Throw an exception if there's already a record for the current path
    if (getRecordForPath(path) != null) {
      throw new KeeperException.NodeExistsException(path);
    }
  } catch (KeeperException.NoNodeException e) {
  }
  // Check if it's an ephemeral parent
  boolean ephemeralParent =
    EphemeralType.get(parentRecord.stat.getEphemeralOwner()) == EphemeralType.NORMAL;
  if (ephemeralParent) {
    throw new KeeperException.NoChildrenForEphemeralsException(path);
  }
  // Calculate new creation version number
  int newCversion = parentRecord.stat.getCversion() + 1;
  // If the operation type is creating a container, set txn to CreateContainerTxn
  if (type == OpCode.createContainer) {
    request.setTxn(new CreateContainerTxn(path, data, listACL, newCversion));
  }
  // If the operation type is creating an expiring node, set txn to CreateTTLTxn
  else if (type == OpCode.createTTL) {
    request.setTxn(new CreateTTLTxn(path, data, listACL, newCversion, ttl));
  }
  // For other creation-related operation types, set txn to CreateTxn
  else {
    request.setTxn(new CreateTxn(path, data, listACL, createMode.isEphemeral(),
                                 newCversion));
  }
  StatPersisted s = new StatPersisted();
  if (createMode.isEphemeral()) {
    s.setEphemeralOwner(request.sessionId);
  }
  // Copy parent node data
  parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
  // Increment the number of child nodes in the parent node
  parentRecord.childCount++;
  // Set creation version number
  parentRecord.stat.setCversion(newCversion);
  // Add change record
  addChangeRecord(parentRecord);
  addChangeRecord(new ChangeRecord(request.getHdr().getZxid(), path, s, 0, listACL));
}

case OpCode.deleteContainer: {
  // Extract the path from the request
  String path = new String(request.request.array());
  // Extract the parent path
  String parentPath = getParentPathAndValidate(path);
  // Parent node record information
  ChangeRecord parentRecord = getRecordForPath(parentPath);
  // Current node record information
  ChangeRecord nodeRecord = getRecordForPath(path);
  // Throw an exception if the current node has child nodes
  if (nodeRecord.childCount > 0) {
    throw new KeeperException.NotEmptyException(path);
  }
  if (EphemeralType.get(nodeRecord.stat.getEphemeralOwner())
      == EphemeralType.NORMAL) {
    throw new KeeperException.BadVersionException(path);
  }
  // Set txn
  request.setTxn(new DeleteTxn(path));
  // Copy parent node data
  parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
  // Decrease the child count of the parent node by one
  parentRecord.childCount--;
  // Add change records
  addChangeRecord(parentRecord);
  addChangeRecord(new ChangeRecord(request.getHdr().getZxid(), path, null, -1, null));
  break;
}

case OpCode.delete:
  // Validate session using sessionTracker
  zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
  // Convert to delete request
  DeleteRequest deleteRequest = (DeleteRequest) record;
  // Deserialize information from the request object into the delete request if needed
  if (deserialize) {
    ByteBufferInputStream.byteBuffer2Record(request.request, deleteRequest);
  }
  // Get the path to be deleted
  String path = deleteRequest.getPath();
  // Get the parent path of the delete path
  String parentPath = getParentPathAndValidate(path);
  // Get the record of the parent path
  ChangeRecord parentRecord = getRecordForPath(parentPath);
  // Get the record of the current path
  ChangeRecord nodeRecord = getRecordForPath(path);
  // Verify ACL
  checkACL(zks, parentRecord.acl, ZooDefs.Perms.DELETE, request.authInfo);
  // Verify version number
  checkAndIncVersion(nodeRecord.stat.getVersion(), deleteRequest.getVersion(), path);
  // Throw an exception if the current node has child nodes
  if (nodeRecord.childCount > 0) {
    throw new KeeperException.NotEmptyException(path);
  }
  // Set txn
  request.setTxn(new DeleteTxn(path));
  // Copy parent node
  parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
  // Decrease the child count of the parent node by one
  parentRecord.childCount--;
  // Add change records
  addChangeRecord(parentRecord);
  addChangeRecord(new ChangeRecord(request.getHdr().getZxid(), path, null, -1, null));
  break;

private static int checkAndIncVersion(int currentVersion, int expectedVersion, String path)
  throws KeeperException.BadVersionException {
  if (expectedVersion != -1 && expectedVersion != currentVersion) {
    throw new KeeperException.BadVersionException(path);
  }
  return currentVersion + 1;
}

case OpCode.setData: 
  // Validate session using sessionTracker
  zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
  // Convert to SetDataRequest
  SetDataRequest setDataRequest = (SetDataRequest) record;
  // Deserialize request object information into SetDataRequest if needed
  if (deserialize) {
    ByteBufferInputStream.byteBuffer2Record(request.request, setDataRequest);
  }
  // Get current operation path
  path = setDataRequest.getPath();
  // Validate current operation path
  validatePath(path, request.sessionId);
  // Get record for current operation path
  nodeRecord = getRecordForPath(path);
  // Verify permissions
  checkACL(zks, nodeRecord.acl, ZooDefs.Perms.WRITE, request.authInfo);
  // Calculate new version
  int newVersion = checkAndIncVersion(nodeRecord.stat.getVersion(),
                                      setDataRequest.getVersion(), path);

  // Set transaction
  request.setTxn(new SetDataTxn(path, setDataRequest.getData(), newVersion));
  // Duplicate current node's record
  nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());
  // Set version number
  nodeRecord.stat.setVersion(newVersion);
  // Add change record
  addChangeRecord(nodeRecord);
  break;

case OpCode.reconfig:
  // Throw exception if configuration reload is not enabled in ZK service
  if (!zks.isReconfigEnabled()) {
    LOG.error("Reconfig operation requested but reconfig feature is disabled.");
    throw new KeeperException.ReconfigDisabledException();
  }

  // Output warning log if skipping ACL
  if (skipACL) {
    LOG.warn("skipACL is set, reconfig operation will skip ACL checks!");
  }

  // Validate session using sessionTracker
  zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
  // Convert to ReconfigRequest
  ReconfigRequest reconfigRequest = (ReconfigRequest) record;
  // Confirm leader service
  LeaderZooKeeperServer lzks;
  try {
    // Throw exception if current ZK service cannot be converted to LeaderZooKeeperServer
    lzks = (LeaderZooKeeperServer) zks;
  } catch (ClassCastException e) {
    throw new KeeperException.UnimplementedException();
  }
  // Get last seen quorum verifier
  QuorumVerifier lastSeenQV = lzks.self.getLastSeenQuorumVerifier();
  // Throw exception if last seen verifier version doesn't match leader service verifier version
  if (lastSeenQV.getVersion() != lzks.self.getQuorumVerifier().getVersion()) {
    throw new KeeperException.ReconfigInProgress();
  }
  // Get configuration ID
  long configId = reconfigRequest.getCurConfigId();

  // Configuration ID is -1
  // Configuration ID doesn't match last seen verifier version
  if (configId != -1 && configId != lzks.self.getLastSeenQuorumVerifier()
      .getVersion()) {
    String msg = "Reconfiguration from version " + configId
      + " failed -- last seen version is " +
      lzks.self.getLastSeenQuorumVerifier().getVersion();
    throw new KeeperException.BadVersionException(msg);
  }

  // Get new members (configuration data)
  String newMembers = reconfigRequest.getNewMembers();

  // New members not empty
  if (newMembers != null) { 
    LOG.info("Non-incremental reconfig");
    newMembers = newMembers.replaceAll(",", "\n");

    try {
      // Read new member data
      Properties props = new Properties();
      props.load(new StringReader(newMembers));
      // Parse configuration
      request.qv = QuorumPeerConfig.parseDynamicConfig(props,
                                                       lzks.self.getElectionType(), true, false);
      request.qv.setVersion(request.getHdr().getZxid());
    } catch (IOException | ConfigException e) {
      throw new KeeperException.BadArgumentsException(e.getMessage());
    }
  } else {
    LOG.info("Incremental reconfig");

    // Determine servers joining the cluster
    List<String> joiningServers = null;
    String joiningServersString = reconfigRequest.getJoiningServers();
    if (joiningServersString != null) {
      joiningServers = StringUtils.split(joiningServersString, ",");
    }

    // Determine servers leaving the cluster
    List<String> leavingServers = null;
    String leavingServersString = reconfigRequest.getLeavingServers();
    if (leavingServersString != null) {
      leavingServers = StringUtils.split(leavingServersString, ",");
    }

    // Throw exception if types differ
    if (!(lastSeenQV instanceof QuorumMaj)) {
      String msg =
        "Incremental reconfiguration requested but last configuration seen has a non-majority quorum system";
      LOG.warn(msg);
      throw new KeeperException.BadArgumentsException(msg);
    }
    // Server collection
    Map<Long, QuorumServer> nextServers =
      new HashMap<Long, QuorumServer>(lastSeenQV.getAllMembers());
    try {
      // Remove corresponding data from nextServers if leaving servers exist
      if (leavingServers != null) {
        for (String leaving : leavingServers) {
          long sid = Long.parseLong(leaving);
          nextServers.remove(sid);
        }
      }
      // Add data to nextServers if joining servers exist
      if (joiningServers != null) {
        for (String joiner : joiningServers) {
          String[] parts =
            StringUtils.split(joiner, "=").toArray(new String[0]);
          if (parts.length != 2) {
            throw new KeeperException.BadArgumentsException(
              "Wrong format of server string");
          }
          Long sid = Long.parseLong(
            parts[0].substring(parts[0].lastIndexOf('.') + 1));
          QuorumServer qs = new QuorumServer(sid, parts[1]);
          if (qs.clientAddr == null || qs.electionAddr == null
              || qs.addr == null) {
            throw new KeeperException.BadArgumentsException(
              "Wrong format of server string - each server should have 3 ports specified");
          }
          for (QuorumServer nqs : nextServers.values()) {
            if (qs.id == nqs.id) {
              continue;
            }
            qs.checkAddressDuplicate(nqs);
          }

          nextServers.remove(qs.id);
          nextServers.put(qs.id, qs);
        }
      }
    } catch (ConfigException e) {
      throw new KeeperException.BadArgumentsException("Reconfiguration failed");
    }
    // Set request parameter qv
    request.qv = new QuorumMaj(nextServers);
    request.qv.setVersion(request.getHdr().getZxid());
  }
  // Throw exception if quorum configuration is enabled and voting members are less than 2
  if (QuorumPeerConfig.isStandaloneEnabled()
      && request.qv.getVotingMembers().size() < 2) {
    String msg =
      "Reconfig failed - new configuration must include at least 2 followers";
    LOG.warn(msg);
    throw new KeeperException.BadArgumentsException(msg);
  }
  // Throw exception if voting members are less than 1
  else if (request.qv.getVotingMembers().size() < 1) {
    String msg =
      "Reconfig failed - new configuration must include at least 1 follower";
    LOG.warn(msg);
    throw new KeeperException.BadArgumentsException(msg);
  }

  // Check if quorum data is synced, throw exception if not
  if (!lzks.getLeader().isQuorumSynced(request.qv)) {
    String msg2 =
      "Reconfig failed - there must be a connected and synced quorum in new configuration";
    LOG.warn(msg2);
    throw new KeeperException.NewConfigNoQuorum();
  }

  // Get record for configuration node
  nodeRecord = getRecordForPath(ZooDefs.CONFIG_NODE);
  // Verify permissions
  checkACL(zks, nodeRecord.acl, ZooDefs.Perms.WRITE, request.authInfo);
  // Set transaction
  request.setTxn(
    new SetDataTxn(ZooDefs.CONFIG_NODE, request.qv.toString().getBytes(), -1));
  nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());
  nodeRecord.stat.setVersion(-1);
  addChangeRecord(nodeRecord);
  break;

case OpCode.setACL:
  // Validate session using sessionTracker
  zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
  // Type conversion
  SetACLRequest setAclRequest = (SetACLRequest) record;
  // Deserialization
  if (deserialize) {
    ByteBufferInputStream.byteBuffer2Record(request.request, setAclRequest);
  }
  // Get operation path
  path = setAclRequest.getPath();
  // Validate path
  validatePath(path, request.sessionId);
  // Calculate ACL
  List<ACL> listACL = fixupACL(path, request.authInfo, setAclRequest.getAcl());
  // Get record for current path
  nodeRecord = getRecordForPath(path);
  // Verify ACL
  checkACL(zks, nodeRecord.acl, ZooDefs.Perms.ADMIN, request.authInfo);
  // Calculate new version number
  newVersion = checkAndIncVersion(nodeRecord.stat.getAversion(),
                                  setAclRequest.getVersion(), path);
  // Set transaction
  request.setTxn(new SetACLTxn(path, listACL, newVersion));
  // Duplicate record for current path
  nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());
  nodeRecord.stat.setAversion(newVersion);
  addChangeRecord(nodeRecord);
  break;

case OpCode.createSession:
  request.request.rewind();
  // Timeout
  int to = request.request.getInt();
  request.setTxn(new CreateSessionTxn(to));
  request.request.rewind();
  // Set sessionId for different states
  if (request.isLocalSession()) {
    zks.sessionTracker.addSession(request.sessionId, to);
  } else {
    zks.sessionTracker.addGlobalSession(request.sessionId, to);
  }
  zks.setOwner(request.sessionId, request.getOwner());
  break;

case OpCode.closeSession:
  // Get the corresponding ephemeral paths for the sessionId
  Set<String> es = zks.getZKDatabase()
    .getEphemerals(request.sessionId);
  // Lock
  synchronized (zks.outstandingChanges) {
    // Iterate through outstandingChanges object
    for (ChangeRecord c : zks.outstandingChanges) {
      // If the status is null
      if (c.stat == null) {
        // Performing a delete
        // Remove from the ephemeral path set
        es.remove(c.path);
      }
      // If the ephemeral owner of the status matches the current sessionId, add it to the ephemeral path set
      else if (c.stat.getEphemeralOwner() == request.sessionId) {
        es.add(c.path);
      }
    }
    // Iterate to mark for deletion
    for (String path2Delete : es) {
      addChangeRecord(
        new ChangeRecord(request.getHdr().getZxid(), path2Delete, null, 0,
                         null));
    }

    // Close the sessionId
    zks.sessionTracker.setSessionClosing(request.sessionId);
  }
  break;

case OpCode.check:
  // Validate session using sessionTracker
  zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
  // Type conversion
  CheckVersionRequest checkVersionRequest = (CheckVersionRequest) record;
  // Deserialization
  if (deserialize) {
    ByteBufferInputStream.byteBuffer2Record(request.request, checkVersionRequest);
  }
  // Extract the operation path
  path = checkVersionRequest.getPath();
  // Validate path
  validatePath(path, request.sessionId);
  // Get the record for the operation path
  nodeRecord = getRecordForPath(path);
  // Verify ACL
  checkACL(zks, nodeRecord.acl, ZooDefs.Perms.READ, request.authInfo);
  // Set txn
  request.setTxn(
    new CheckVersionTxn(path, checkAndIncVersion(nodeRecord.stat.getVersion(),
                                                 checkVersionRequest.getVersion(), path)));
  break;

case OpCode.multi:{
  // Create a multi-transaction request object
  MultiTransactionRecord multiRequest = new MultiTransactionRecord();
  try {
    // Convert request object to multi-transaction request object
    ByteBufferInputStream.byteBuffer2Record(request.request, multiRequest);
  } catch (IOException e) {
    // Set header information for exception
    request.setHdr(
      new TxnHeader(request.sessionId, request.cxid, zks.getNextZxid(),
                    Time.currentWallTime(), OpCode.multi));
    throw e;
  }
  List<Txn> txns = new ArrayList<Txn>();
  // Get next zxid
  long zxid = zks.getNextZxid();
  // Exception record variable
  KeeperException ke = null;
  // Convert multi-transaction request object to map structure for subsequent rollback operations
  Map<String, ChangeRecord> pendingChanges = getPendingChanges(multiRequest);

  // Iterate through multi-transaction request object
  for (Op op : multiRequest) {
    // Get request information from single transaction
    Record subrequest = op.toRequestRecord();
    // Operation type
    int type;
    // Record information
    Record txn;
    // If exception record variable is not null, set exception-related information
    if (ke != null) {
      type = OpCode.error;
      txn = new ErrorTxn(Code.RUNTIMEINCONSISTENCY.intValue());
    }
    else {
      try {
        // Execute core request-to-transaction logic
        pRequest2Txn(op.getType(), zxid, request, subrequest, false);
        // Record operation type
        type = request.getHdr().getType();
        // Record txn
        txn = request.getTxn();
      }
      // Record exception-related information
      catch (KeeperException e) {

        ke = e;
        type = OpCode.error;
        txn = new ErrorTxn(e.code().intValue());

        if (e.code().intValue() > Code.APIERROR.intValue()) {
          LOG.info(
            "Got user-level KeeperException when processing {} aborting"
            +
            " remaining multi ops. Error Path:{} Error:{}",
            request.toString(), e.getPath(), e.getMessage());
        }

        request.setException(e);
        // Rollback operation
        rollbackPendingChanges(zxid, pendingChanges);
      }
    }
    // Output record processing
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
    txn.serialize(boa, "request");
    ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());

    txns.add(new Txn(type, bb.array()));
  }

  request.setHdr(new TxnHeader(request.sessionId, request.cxid, zxid,
                               Time.currentWallTime(), request.type));
  request.setTxn(new MultiTxn(txns));

  break;
}

void rollbackPendingChanges(long zxid, Map<String, ChangeRecord> pendingChangeRecords) {
  // Lock
  synchronized (zks.outstandingChanges) {
    // Iterate through outstandingChanges variable
    Iterator<ChangeRecord> iter = zks.outstandingChanges.descendingIterator();
    while (iter.hasNext()) {
      ChangeRecord c = iter.next();
      // If the operation record matches the current record
      if (c.zxid == zxid) {
        iter.remove();
        zks.outstandingChangesForPath.remove(c.path);
      }
      else {
        break;
      }
    }

    // If outstandingChanges variable is empty
    if (zks.outstandingChanges.isEmpty()) {
      return;
    }

    // Get the first zxid
    long firstZxid = zks.outstandingChanges.peek().zxid;

    // Skip processing for those less than the first zxid, add those greater than or equal to outstandingChangesForPath variable
    for (ChangeRecord c : pendingChangeRecords.values()) {
      if (c.zxid < firstZxid) {
        continue;
      }
      zks.outstandingChangesForPath.put(c.path, c);
    }
  }
}

public LeaderRequestProcessor(LeaderZooKeeperServer zks,
                              RequestProcessor nextProcessor) {
  this.lzks = zks;
  this.nextProcessor = nextProcessor;
}

@Override
protected void setupRequestProcessors() {
  RequestProcessor finalProcessor = new FinalRequestProcessor(this);
  RequestProcessor toBeAppliedProcessor =
    new Leader.ToBeAppliedRequestProcessor(finalProcessor, getLeader());
  commitProcessor = new CommitProcessor(toBeAppliedProcessor,
                                        Long.toString(getServerId()), false,
                                        getZooKeeperServerListener());
  commitProcessor.start();
  ProposalRequestProcessor proposalProcessor = new ProposalRequestProcessor(this,
                                                                            commitProcessor);
  proposalProcessor.initialize();
  prepRequestProcessor = new PrepRequestProcessor(this, proposalProcessor);
  prepRequestProcessor.start();
  firstProcessor = new LeaderRequestProcessor(this, prepRequestProcessor);

  setupContainerManager();
}

@Override
public void processRequest(Request request)
  throws RequestProcessorException {
  // Upgraded request object
  Request upgradeRequest = null;
  try {
    // Validate session to determine if an upgrade is needed, and perform the upgrade if necessary
    upgradeRequest = lzks.checkUpgradeSession(request);
  } catch (KeeperException ke) {
    if (request.getHdr() != null) {
      LOG.debug("Updating header");
      request.getHdr().setType(OpCode.error);
      request.setTxn(new ErrorTxn(ke.code().intValue()));
    }
    request.setException(ke);
    LOG.info("Error creating upgrade request " + ke.getMessage());
  } catch (IOException ie) {
    LOG.error("Unexpected error in upgrade", ie);
  }
  // If the upgraded request object is not null
  if (upgradeRequest != null) {
    nextProcessor.processRequest(upgradeRequest);
  }
  nextProcessor.processRequest(request);
}

@Override
public void processRequest(Request request) {
  if (stopped) {
    return;
  }
  if (LOG.isDebugEnabled()) {
    LOG.debug("Processing request:: " + request);
  }
  queuedRequests.add(request);
  if (!isWaitingForCommit()) {
    wakeup();
  }
}

@Override
public void run() {
  Request request;
  try {
    // Loop while not stopped
    while (!stopped) {
      // Lock
      synchronized (this) {
        // Conditions for waiting:
        // 1. Not stopped
        // 2.1 queuedRequests set is empty
        // 2.2 nextPending contains data
        // 2.3 currentlyCommitting contains data
        // 3.1 committedRequests is empty
        // 3.2 numRequestsProcessing count is not 0
        while (
          !stopped &&
          ((queuedRequests.isEmpty() || isWaitingForCommit()
            || isProcessingCommit()) &&
           (committedRequests.isEmpty()
            || isProcessingRequest()))) {

          wait();
        }
      }
      // Process requests in queuedRequests
      // Loop conditions:
      // 1. Not stopped working
      // 2. nextPending contains data
      // 3. currentlyCommitting contains data
      // 4. Get a request object from queuedRequests, and this object is not null
      while (!stopped && !isWaitingForCommit() &&
             !isProcessingCommit() &&
             (request = queuedRequests.poll()) != null) {
        // Determine if commitment is needed
        if (needCommit(request)) {
          nextPending.set(request);
        }
        else {
          // Send to the next request processor for processing
          sendToNextProcessor(request);
        }
      }
      // Process committed requests
      processCommitted();
    }
  } catch (Throwable e) {
    handleException(this.getName(), e);
  }
  LOG.info("CommitProcessor exited loop!");
}

private void sendToNextProcessor(Request request) {
  numRequestsProcessing.incrementAndGet();
  workerPool.schedule(new CommitWorkRequest(request), request.sessionId);
}

public void schedule(WorkRequest workRequest, long id) {
  // Check if processing is stopped
  if (stopped) {
    // Clean up the request
    workRequest.cleanup();
    return;
  }

  // Convert the request
  ScheduledWorkRequest scheduledWorkRequest =
    new ScheduledWorkRequest(workRequest);
  // Get the number of task executors
  int size = workers.size();
  // If there are task executors available
  if (size > 0) {
    try {
      int workerNum = ((int) (id % size) + size) % size;
      // Get a task executor and execute
      ExecutorService worker = workers.get(workerNum);
      worker.execute(scheduledWorkRequest);
    } catch (RejectedExecutionException e) {
      LOG.warn("ExecutorService rejected execution", e);
      workRequest.cleanup();
    }
  }
  else {
    scheduledWorkRequest.run();
  }
}

private class ScheduledWorkRequest implements Runnable {
  private final WorkRequest workRequest;

  ScheduledWorkRequest(WorkRequest workRequest) {
    this.workRequest = workRequest;
  }

  @Override
  public void run() {
    try {
      if (stopped) {
        workRequest.cleanup();
        return;
      }
      workRequest.doWork();
    } catch (Exception e) {
      LOG.warn("Unexpected exception", e);
      workRequest.cleanup();
    }
  }
}

public void doWork() throws RequestProcessorException {
    try {
        nextProcessor.processRequest(request);
    } finally {
        currentlyCommitting.compareAndSet(request, null);
        if (numRequestsProcessing.decrementAndGet() == 0) {
            if (!queuedRequests.isEmpty() ||
                    !committedRequests.isEmpty()) {
                wakeup();
            }
        }
    }
}

protected void processCommitted() {
  // Request object
  Request request;

  // Execute if conditions are met
  // 1. Work has not stopped
  // 2. numRequestsProcessing count is 0
  // 3. Data retrieved from committedRequests is not null
  if (!stopped && !isProcessingRequest() &&
      (committedRequests.peek() != null)) {
    // 1. Data exists in nextPending
    // 2. queuedRequests is not empty
    if (!isWaitingForCommit() && !queuedRequests.isEmpty()) {
      return;
    }
    // Retrieve and process the request
    request = committedRequests.poll();
    // Get the next request from nextPending
    Request pending = nextPending.get();
    // 1. The next request is not null
    // 2. The sessionId of the next request matches the current request
    // 3. The cxid of the next request matches the current request
    if (pending != null &&
        pending.sessionId == request.sessionId &&
        pending.cxid == request.cxid) {
      pending.setHdr(request.getHdr());
      pending.setTxn(request.getTxn());
      pending.zxid = request.zxid;
      currentlyCommitting.set(pending);
      nextPending.set(null);
      sendToNextProcessor(pending);
    }
    else {
      currentlyCommitting.set(request);
      sendToNextProcessor(request);
    }
  }
}

public void processRequest(Request request) throws RequestProcessorException {
  if (request instanceof LearnerSyncRequest) {
    zks.getLeader().processSync((LearnerSyncRequest) request);
  } else {
    nextProcessor.processRequest(request);
    if (request.getHdr() != null) {
      try {
        // Send to other services
        zks.getLeader().propose(request);
      } catch (XidRolloverException e) {
        throw new RequestProcessorException(e.getMessage(), e);
      }
      syncProcessor.processRequest(request);
    }
  }
}

@Override
public void run() {
  try {
    // Log counter
    int logCount = 0;
    // Random snapshot count to avoid full snapshot logging
    int randRoll = r.nextInt(snapCount / 2);
    while (true) {
      Request si = null;
      // Retrieve request object from queuedRequests collection
      if (toFlush.isEmpty()) {
        si = queuedRequests.take();
      } else {
        si = queuedRequests.poll();
        if (si == null) {
          flush(toFlush);
          continue;
        }
      }
      // Break the loop if the request object matches the death request
      if (si == requestOfDeath) {
        break;
      }
      if (si != null) {
        // Append current request to ZooKeeper database
        if (zks.getZKDatabase().append(si)) {
          // Increment log counter
          logCount++;
          // Check log counter
          if (logCount > (snapCount / 2 + randRoll)) {
            randRoll = r.nextInt(snapCount / 2);
            // Roll the log
            zks.getZKDatabase().rollLog();
            // Check if snapshot generation is needed
            if (snapInProcess != null && snapInProcess.isAlive()) {
              LOG.warn("Too busy to snap, skipping");
            }
            else {
              snapInProcess = new ZooKeeperThread("Snapshot Thread") {
                public void run() {
                  try {
                    // Generate snapshot
                    zks.takeSnapshot();
                  } catch (Exception e) {
                    LOG.warn("Unexpected exception", e);
                  }
                }
              };
              // Start the thread
              snapInProcess.start();
            }
            // Reset counter
            logCount = 0;
          }
        }
        // If toFlush is empty
        else if (toFlush.isEmpty()) {
          if (nextProcessor != null) {
            nextProcessor.processRequest(si);
            if (nextProcessor instanceof Flushable) {
              ((Flushable) nextProcessor).flush();
            }
          }
          continue;
        }
        // Add to pending flush requests
        toFlush.add(si);
        // Perform write operation if over 1000 requests
        if (toFlush.size() > 1000) {
          flush(toFlush);
        }
      }
    }
  } catch (Throwable t) {
    handleException(this.getName(), t);
  } finally {
    running = false;
  }
  LOG.info("SyncRequestProcessor exited!");
}