Java 类org.apache.zookeeper.server.Request 实例源码

项目:fuck_zookeeper    文件:FollowerZooKeeperServer.java   
/**
 * When a COMMIT message is received, eventually this method is called, 
 * which matches up the zxid from the COMMIT with (hopefully) the head of
 * the pendingTxns queue and hands it to the commitProcessor to commit.
 * @param zxid - must correspond to the head of pendingTxns if it exists
 */
public void commit(long zxid) {
    if (pendingTxns.size() == 0) {
        LOG.warn("Committing " + Long.toHexString(zxid)
                + " without seeing txn");
        return;
    }
    long firstElementZxid = pendingTxns.element().zxid;
    if (firstElementZxid != zxid) {
        LOG.error("Committing zxid 0x" + Long.toHexString(zxid)
                + " but next pending txn 0x"
                + Long.toHexString(firstElementZxid));
        System.exit(12);
    }
    Request request = pendingTxns.remove();
    commitProcessor.commit(request);
}
项目:fuck_zookeeper    文件:SendAckRequestProcessor.java   
public void processRequest(Request si) {
    if(si.type != OpCode.sync){
        QuorumPacket qp = new QuorumPacket(Leader.ACK, si.hdr.getZxid(), null,
            null);
        try {
            learner.writePacket(qp, false);
        } catch (IOException e) {
            LOG.warn("Closing connection to leader, exception during packet send", e);
            try {
                if (!learner.sock.isClosed()) {
                    learner.sock.close();
                }
            } catch (IOException e1) {
                // Nothing to do, we are shutting things down, so an exception here is irrelevant
                LOG.debug("Ignoring error closing the connection", e1);
            }
        }
    }
}
项目:https-github.com-apache-zookeeper    文件:FollowerRequestProcessor.java   
public void processRequest(Request request) {
    if (!finished) {
        // Before sending the request, check if the request requires a
        // global session and what we have is a local session. If so do
        // an upgrade.
        Request upgradeRequest = null;
        try {
            upgradeRequest = zks.checkUpgradeSession(request);
        } catch (KeeperException ke) {
            if (request.getHdr() != null) {
                request.getHdr().setType(OpCode.error);
                request.setTxn(new ErrorTxn(ke.code().intValue()));
            }
            request.setException(ke);
            LOG.info("Error creating upgrade request",  ke);
        } catch (IOException ie) {
            LOG.error("Unexpected error in upgrade", ie);
        }
        if (upgradeRequest != null) {
            queuedRequests.add(upgradeRequest);
        }
        queuedRequests.add(request);
    }
}
项目:https-github.com-apache-zookeeper    文件:Leader.java   
public void processRequest(Request request) throws RequestProcessorException {
    next.processRequest(request);

    // The only requests that should be on toBeApplied are write
    // requests, for which we will have a hdr. We can't simply use
    // request.zxid here because that is set on read requests to equal
    // the zxid of the last write op.
    if (request.getHdr() != null) {
        long zxid = request.getHdr().getZxid();
        Iterator<Proposal> iter = leader.toBeApplied.iterator();
        if (iter.hasNext()) {
            Proposal p = iter.next();
            if (p.request != null && p.request.zxid == zxid) {
                iter.remove();
                return;
            }
        }
        LOG.error("Committed request not found on toBeApplied: "
                  + request);
    }
}
项目:https-github.com-apache-zookeeper    文件:FollowerZooKeeperServer.java   
/**
 * When a COMMIT message is received, eventually this method is called,
 * which matches up the zxid from the COMMIT with (hopefully) the head of
 * the pendingTxns queue and hands it to the commitProcessor to commit.
 * @param zxid - must correspond to the head of pendingTxns if it exists
 */
public void commit(long zxid) {
    if (pendingTxns.size() == 0) {
        LOG.warn("Committing " + Long.toHexString(zxid)
                + " without seeing txn");
        return;
    }
    long firstElementZxid = pendingTxns.element().zxid;
    if (firstElementZxid != zxid) {
        LOG.error("Committing zxid 0x" + Long.toHexString(zxid)
                + " but next pending txn 0x"
                + Long.toHexString(firstElementZxid));
        System.exit(12);
    }
    Request request = pendingTxns.remove();
    commitProcessor.commit(request);
}
项目:https-github.com-apache-zookeeper    文件:ObserverRequestProcessor.java   
/**
 * Simply queue the request, which will be processed in FIFO order.
 */
public void processRequest(Request request) {
    if (!finished) {
        Request upgradeRequest = null;
        try {
            upgradeRequest = zks.checkUpgradeSession(request);
        } catch (KeeperException ke) {
            if (request.getHdr() != null) {
                request.getHdr().setType(OpCode.error);
                request.setTxn(new ErrorTxn(ke.code().intValue()));
            }
            request.setException(ke);
            LOG.info("Error creating upgrade request",  ke);
        } catch (IOException ie) {
            LOG.error("Unexpected error in upgrade", ie);
        }
        if (upgradeRequest != null) {
            queuedRequests.add(upgradeRequest);
        }
        queuedRequests.add(request);
    }
}
项目:https-github.com-apache-zookeeper    文件:LeaderZooKeeperServer.java   
/**
 * Requests coming from the learner should go directly to
 * PrepRequestProcessor
 *
 * @param request
 */
public void submitLearnerRequest(Request request) {
    /*
     * Requests coming from the learner should have gone through
     * submitRequest() on each server which already perform some request
     * validation, so we don't need to do it again.
     *
     * Additionally, LearnerHandler should start submitting requests into
     * the leader's pipeline only when the leader's server is started, so we
     * can submit the request directly into PrepRequestProcessor.
     *
     * This is done so that requests from learners won't go through
     * LeaderRequestProcessor which perform local session upgrade.
     */
    prepRequestProcessor.processRequest(request);
}
项目:https-github.com-apache-zookeeper    文件:CommitProcessor.java   
protected boolean needCommit(Request request) {
    switch (request.type) {
        case OpCode.create:
        case OpCode.create2:
        case OpCode.createTTL:
        case OpCode.createContainer:
        case OpCode.delete:
        case OpCode.deleteContainer:
        case OpCode.setData:
        case OpCode.reconfig:
        case OpCode.multi:
        case OpCode.setACL:
            return true;
        case OpCode.sync:
            return matchSyncs;    
        case OpCode.createSession:
        case OpCode.closeSession:
            return !request.isLocalSession();
        default:
            return false;
    }
}
项目:https-github.com-apache-zookeeper    文件:LeaderRequestProcessor.java   
@Override
public void processRequest(Request request)
        throws RequestProcessorException {
    // Check if this is a local session and we are trying to create
    // an ephemeral node, in which case we upgrade the session
    Request upgradeRequest = null;
    try {
        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 (upgradeRequest != null) {
        nextProcessor.processRequest(upgradeRequest);
    }

    nextProcessor.processRequest(request);
}
项目:https-github.com-apache-zookeeper    文件:QuorumZooKeeperServer.java   
@Override
protected void setLocalSessionFlag(Request si) {
    // We need to set isLocalSession to tree for these type of request
    // so that the request processor can process them correctly.
    switch (si.type) {
    case OpCode.createSession:
        if (self.areLocalSessionsEnabled()) {
            // All new sessions local by default.
            si.setLocalSession(true);
        }
        break;
    case OpCode.closeSession:
        String reqType = "global";
        if (upgradeableSessionTracker.isLocalSession(si.sessionId)) {
            si.setLocalSession(true);
            reqType = "local";
        }
        LOG.info("Submitting " + reqType + " closeSession request"
                + " for session 0x" + Long.toHexString(si.sessionId));
        break;
    default:
        break;
    }
}
项目:https-github.com-apache-zookeeper    文件:CommitProcessorTest.java   
@Override
public void run() {
    Random rand = new Random(Thread.currentThread().getId());
    try {
        while(true) {
            // If it is a read-only test, there will be no proposals..
            if (!proposals.isEmpty()){
                Request request = proposals.take();
                Thread.sleep(5 + rand.nextInt(95));
                commitProcessor.commit(request);
            }
        }
    } catch (InterruptedException e) {
        // ignore
    }
}
项目:https-github.com-apache-zookeeper    文件:CommitProcessorTest.java   
/**
 * Validate that this is the only request in the pipeline
 */
private void validateWriteRequestVariant(Request request) {
    if (stopped)
        return;
    long zxid = request.getHdr().getZxid();
    int readRequests = outstandingReadRequests.get();
    if (readRequests != 0) {
        failTest("There are " + readRequests + " outstanding"
                + " read requests while issuing a write request zxid="
                + zxid);
    }
    int writeRequests = outstandingWriteRequests.get();
    if (writeRequests > 1) {
        failTest("There are " + writeRequests + " outstanding"
                + " write requests while issuing a write request zxid="
                + zxid + " (expected one)");
    }
}
项目:https-github.com-apache-zookeeper    文件:CommitProcessorConcurrencyTest.java   
MockCommitProcessor() {
    super(new RequestProcessor() {
        public void processRequest(Request request)
                throws RequestProcessorException {
            processedRequests.offer(request);
        }

        public void shutdown() {
        }
    }, "0", false, new ZooKeeperServerListener() {

        @Override
        public void notifyStopping(String threadName, int errorCode) {
            Assert.fail("Commit processor crashed " + errorCode);
        }
    });
}
项目:https-github.com-apache-zookeeper    文件:CommitProcessorConcurrencyTest.java   
/**
 * In the following test, we verify that committed requests are processed
 * even when queuedRequests never gets empty. We add 10 committed request
 * and use infinite queuedRequests. We verify that the committed request was
 * processed.
 */
@Test(timeout = 1000)
public void noStarvationOfNonLocalCommittedRequestsTest() throws Exception {
    final String path = "/noStarvationOfCommittedRequests";
    processor.queuedRequests = new MockRequestsQueue();
    Set<Request> nonLocalCommits = new HashSet<Request>();
    for (int i = 0; i < 10; i++) {
        Request nonLocalCommitReq = newRequest(
                new CreateRequest(path, new byte[0], Ids.OPEN_ACL_UNSAFE,
                        CreateMode.PERSISTENT_SEQUENTIAL.toFlag()),
                OpCode.create, 51, i + 1);
        processor.committedRequests.add(nonLocalCommitReq);
        nonLocalCommits.add(nonLocalCommitReq);
    }
    for (int i = 0; i < 10; i++) {
        processor.initThreads(defaultSizeOfThreadPool);
        processor.stoppedMainLoop = true;
        processor.run();
    }
    Assert.assertTrue("commit request was not processed",
            processedRequests.containsAll(nonLocalCommits));
}
项目:https-github.com-apache-zookeeper    文件:FileTxnSnapLogTest.java   
@Test
public void testGetTxnLogSyncElapsedTime() throws IOException {
    File tmpDir = ClientBase.createEmptyTestDir();
    FileTxnSnapLog fileTxnSnapLog = new FileTxnSnapLog(new File(tmpDir, "data"),
            new File(tmpDir, "data_txnlog"));

    TxnHeader hdr = new TxnHeader(1, 1, 1, 1, ZooDefs.OpCode.setData);
    Record txn = new SetDataTxn("/foo", new byte[0], 1);
    Request req = new Request(0, 0, 0, hdr, txn, 0);

    try {
        fileTxnSnapLog.append(req);
        fileTxnSnapLog.commit();
        long syncElapsedTime = fileTxnSnapLog.getTxnLogElapsedSyncTime();
        Assert.assertNotEquals("Did not update syncElapsedTime!", -1L, syncElapsedTime);
    } finally {
        fileTxnSnapLog.close();
    }
}
项目:ZooKeeper    文件:FollowerZooKeeperServer.java   
/**
 * When a COMMIT message is received, eventually this method is called, 
 * which matches up the zxid from the COMMIT with (hopefully) the head of
 * the pendingTxns queue and hands it to the commitProcessor to commit.
 * @param zxid - must correspond to the head of pendingTxns if it exists
 */
public void commit(long zxid) {
    if (pendingTxns.size() == 0) {
        LOG.warn("Committing " + Long.toHexString(zxid)
                + " without seeing txn");
        return;
    }
    long firstElementZxid = pendingTxns.element().zxid;
    if (firstElementZxid != zxid) {
        LOG.error("Committing zxid 0x" + Long.toHexString(zxid)
                + " but next pending txn 0x"
                + Long.toHexString(firstElementZxid));
        System.exit(12);
    }
    Request request = pendingTxns.remove();
    commitProcessor.commit(request);
}
项目:ZooKeeper    文件:SendAckRequestProcessor.java   
public void processRequest(Request si) {
    if(si.type != OpCode.sync){
        QuorumPacket qp = new QuorumPacket(Leader.ACK, si.hdr.getZxid(), null,
            null);
        try {
            learner.writePacket(qp, false);
        } catch (IOException e) {
            LOG.warn("Closing connection to leader, exception during packet send", e);
            try {
                if (!learner.sock.isClosed()) {
                    learner.sock.close();
                }
            } catch (IOException e1) {
                // Nothing to do, we are shutting things down, so an exception here is irrelevant
                LOG.debug("Ignoring error closing the connection", e1);
            }
        }
    }
}
项目:fuck_zookeeper    文件:FollowerRequestProcessor.java   
public void shutdown() {
    LOG.info("Shutting down");
    finished = true;
    queuedRequests.clear();
    queuedRequests.add(Request.requestOfDeath);
    nextProcessor.shutdown();
}
项目:fuck_zookeeper    文件:ReadOnlyRequestProcessor.java   
@Override
public void shutdown() {
    finished = true;
    queuedRequests.clear();
    queuedRequests.add(Request.requestOfDeath);
    nextProcessor.shutdown();
}
项目:fuck_zookeeper    文件:Leader.java   
public void processRequest(Request request) throws RequestProcessorException {
    // request.addRQRec(">tobe");
    next.processRequest(request);
    Proposal p = toBeApplied.peek();
    if (p != null && p.request != null
            && p.request.zxid == request.zxid) {
        toBeApplied.remove();
    }
}
项目:fuck_zookeeper    文件:FollowerZooKeeperServer.java   
/**
 * @param port
 * @param dataDir
 * @throws IOException
 */
FollowerZooKeeperServer(FileTxnSnapLog logFactory,QuorumPeer self,
        DataTreeBuilder treeBuilder, ZKDatabase zkDb) throws IOException {
    super(logFactory, self.tickTime, self.minSessionTimeout,
            self.maxSessionTimeout, treeBuilder, zkDb, self);
    this.pendingSyncs = new ConcurrentLinkedQueue<Request>();
}
项目:fuck_zookeeper    文件:FollowerZooKeeperServer.java   
public void logRequest(TxnHeader hdr, Record txn) {
    Request request = new Request(null, hdr.getClientId(), hdr.getCxid(),
            hdr.getType(), null, null);
    request.hdr = hdr;
    request.txn = txn;
    request.zxid = hdr.getZxid();
    if ((request.zxid & 0xffffffffL) != 0) {
        pendingTxns.add(request);
    }
    syncProcessor.processRequest(request);
}
项目:fuck_zookeeper    文件:FollowerZooKeeperServer.java   
synchronized public void sync(){
      if(pendingSyncs.size() ==0){
          LOG.warn("Not expecting a sync.");
          return;
      }

      Request r = pendingSyncs.remove();
commitProcessor.commit(r);
  }
项目:fuck_zookeeper    文件:ObserverRequestProcessor.java   
/**
 * Shutdown the processor.
 */
public void shutdown() {
    LOG.info("Shutting down");
    finished = true;
    queuedRequests.clear();
    queuedRequests.add(Request.requestOfDeath);
    nextProcessor.shutdown();
}
项目:fuck_zookeeper    文件:AckRequestProcessor.java   
/**
 * Forward the request as an ACK to the leader
 */
public void processRequest(Request request) {
    QuorumPeer self = leader.self;
    if(self != null)
        leader.processAck(self.getId(), request.zxid, null);
    else
        LOG.error("Null QuorumPeer");
}
项目:fuck_zookeeper    文件:ObserverZooKeeperServer.java   
synchronized public void sync(){
    if(pendingSyncs.size() ==0){
        LOG.warn("Not expecting a sync.");
        return;
    }

    Request r = pendingSyncs.remove();
    commitProcessor.commit(r);
}
项目:fuck_zookeeper    文件:Observer.java   
/**
 * Controls the response of an observer to the receipt of a quorumpacket
 * @param qp
 * @throws IOException
 */
protected void processPacket(QuorumPacket qp) throws IOException{
    switch (qp.getType()) {
    case Leader.PING:
        ping(qp);
        break;
    case Leader.PROPOSAL:
        LOG.warn("Ignoring proposal");
        break;
    case Leader.COMMIT:
        LOG.warn("Ignoring commit");            
        break;            
    case Leader.UPTODATE:
        LOG.error("Received an UPTODATE message after Observer started");
        break;
    case Leader.REVALIDATE:
        revalidate(qp);
        break;
    case Leader.SYNC:
        ((ObserverZooKeeperServer)zk).sync();
        break;
    case Leader.INFORM:            
        TxnHeader hdr = new TxnHeader();
        Record txn = SerializeUtils.deserializeTxn(qp.getData(), hdr);
        Request request = new Request (null, hdr.getClientId(), 
                                       hdr.getCxid(),
                                       hdr.getType(), null, null);
        request.txn = txn;
        request.hdr = hdr;
        ObserverZooKeeperServer obs = (ObserverZooKeeperServer)zk;
        obs.commitRequest(request);            
        break;
    }
}
项目:fuck_zookeeper    文件:CommitProcessor.java   
synchronized public void commit(Request request) {
    if (!finished) {
        if (request == null) {
            LOG.warn("Committed a null!",
                     new Exception("committing a null! "));
            return;
        }
        if (LOG.isDebugEnabled()) {
            LOG.debug("Committing request:: " + request);
        }
        committedRequests.add(request);
        notifyAll();
    }
}
项目:fuck_zookeeper    文件:CommitProcessor.java   
synchronized public void processRequest(Request request) {
    // request.addRQRec(">commit");
    if (LOG.isDebugEnabled()) {
        LOG.debug("Processing request:: " + request);
    }

    if (!finished) {
        queuedRequests.add(request);
        notifyAll();
    }
}
项目:fuck_zookeeper    文件:ProposalRequestProcessor.java   
public void processRequest(Request request) throws RequestProcessorException {
    // LOG.warn("Ack>>> cxid = " + request.cxid + " type = " +
    // request.type + " id = " + request.sessionId);
    // request.addRQRec(">prop");


    /* In the following IF-THEN-ELSE block, we process syncs on the leader. 
     * If the sync is coming from a follower, then the follower
     * handler adds it to syncHandler. Otherwise, if it is a client of
     * the leader that issued the sync command, then syncHandler won't 
     * contain the handler. In this case, we add it to syncHandler, and 
     * call processRequest on the next processor.
     */

    if(request instanceof LearnerSyncRequest){
        zks.getLeader().processSync((LearnerSyncRequest)request);
    } else {
            nextProcessor.processRequest(request);
        if (request.hdr != null) {
            // We need to sync and get consensus on any transactions
            try {
                zks.getLeader().propose(request);
            } catch (XidRolloverException e) {
                throw new RequestProcessorException(e.getMessage(), e);
            }
            syncProcessor.processRequest(request);
        }
    }
}
项目:ZooKeeper    文件:ProposalRequestProcessor.java   
public void processRequest(Request request) throws RequestProcessorException {
    // LOG.warn("Ack>>> cxid = " + request.cxid + " type = " +
    // request.type + " id = " + request.sessionId);
    // request.addRQRec(">prop");


    /* In the following IF-THEN-ELSE block, we process syncs on the leader. 
     * If the sync is coming from a follower, then the follower
     * handler adds it to syncHandler. Otherwise, if it is a client of
     * the leader that issued the sync command, then syncHandler won't 
     * contain the handler. In this case, we add it to syncHandler, and 
     * call processRequest on the next processor.
     */

    if(request instanceof LearnerSyncRequest){
        zks.getLeader().processSync((LearnerSyncRequest)request);
    } else {
            nextProcessor.processRequest(request);
        if (request.hdr != null) {
            // We need to sync and get consensus on any transactions
            try {
                zks.getLeader().propose(request);
            } catch (XidRolloverException e) {
                throw new RequestProcessorException(e.getMessage(), e);
            }
            syncProcessor.processRequest(request);
        }
    }
}
项目:fuck_zookeeper    文件:Zab1_0Test.java   
@Test
public void testInitialAcceptedCurrent() throws Exception {
    File tmpDir = File.createTempFile("test", ".dir", testData);
    tmpDir.delete();
    tmpDir.mkdir();
    try {
        FileTxnSnapLog logFactory = new FileTxnSnapLog(tmpDir, tmpDir);
        File version2 = new File(tmpDir, "version-2");
        version2.mkdir();
        long zxid = ZxidUtils.makeZxid(3, 3);

        TxnHeader hdr = new TxnHeader(1, 1, zxid, 1, ZooDefs.OpCode.error);
        ErrorTxn txn = new ErrorTxn(1);
        byte[] buf = Util.marshallTxnEntry(hdr, txn);
        Request req = new Request(null, 1, 1, ZooDefs.OpCode.error,
                ByteBuffer.wrap(buf), null);
        req.hdr = hdr;
        req.txn = txn;
        logFactory.append(req);
        logFactory.commit();
        ZKDatabase zkDb = new ZKDatabase(logFactory);
        QuorumPeer peer = new QuorumPeer();
        peer.setZKDatabase(zkDb);
        peer.setTxnFactory(logFactory);
        peer.getLastLoggedZxid();
        Assert.assertEquals(3, peer.getAcceptedEpoch());
        Assert.assertEquals(3, peer.getCurrentEpoch());
        Assert.assertEquals(3, Integer
                .parseInt(readContentsOfFile(new File(version2,
                        QuorumPeer.CURRENT_EPOCH_FILENAME))));
        Assert.assertEquals(3, Integer
                .parseInt(readContentsOfFile(new File(version2,
                        QuorumPeer.ACCEPTED_EPOCH_FILENAME))));
    } finally {
        recursiveDelete(tmpDir);
    }
}
项目:fuck_zookeeper    文件:TruncateTest.java   
private void append(ZKDatabase zkdb, int i) throws IOException {
    TxnHeader hdr = new TxnHeader(1, 1, i, 1, ZooDefs.OpCode.setData);
    Record txn = new SetDataTxn("/foo" + i, new byte[0], 1);
    Request req = new Request(null, 0, 0, 0, null, null);
    req.hdr = hdr;
    req.txn = txn;

    zkdb.append(req);
    zkdb.commit();
}
项目:https-github.com-apache-zookeeper    文件:FollowerRequestProcessor.java   
public void shutdown() {
    LOG.info("Shutting down");
    finished = true;
    queuedRequests.clear();
    queuedRequests.add(Request.requestOfDeath);
    nextProcessor.shutdown();
}
项目:https-github.com-apache-zookeeper    文件:ReadOnlyRequestProcessor.java   
@Override
public void shutdown() {
    finished = true;
    queuedRequests.clear();
    queuedRequests.add(Request.requestOfDeath);
    nextProcessor.shutdown();
}
项目:https-github.com-apache-zookeeper    文件:FollowerZooKeeperServer.java   
/**
 * @param port
 * @param dataDir
 * @throws IOException
 */
FollowerZooKeeperServer(FileTxnSnapLog logFactory,QuorumPeer self,
        ZKDatabase zkDb) throws IOException {
    super(logFactory, self.tickTime, self.minSessionTimeout,
            self.maxSessionTimeout, zkDb, self);
    this.pendingSyncs = new ConcurrentLinkedQueue<Request>();
}
项目:https-github.com-apache-zookeeper    文件:FollowerZooKeeperServer.java   
public void logRequest(TxnHeader hdr, Record txn) {
    Request request = new Request(hdr.getClientId(), hdr.getCxid(), hdr.getType(), hdr, txn, hdr.getZxid());
    if ((request.zxid & 0xffffffffL) != 0) {
        pendingTxns.add(request);
    }
    syncProcessor.processRequest(request);
}
项目:https-github.com-apache-zookeeper    文件:FollowerZooKeeperServer.java   
synchronized public void sync(){
      if(pendingSyncs.size() ==0){
          LOG.warn("Not expecting a sync.");
          return;
      }

      Request r = pendingSyncs.remove();
commitProcessor.commit(r);
  }
项目:https-github.com-apache-zookeeper    文件:ObserverRequestProcessor.java   
/**
 * Shutdown the processor.
 */
public void shutdown() {
    LOG.info("Shutting down");
    finished = true;
    queuedRequests.clear();
    queuedRequests.add(Request.requestOfDeath);
    nextProcessor.shutdown();
}
项目:https-github.com-apache-zookeeper    文件:AckRequestProcessor.java   
/**
 * Forward the request as an ACK to the leader
 */
public void processRequest(Request request) {
    QuorumPeer self = leader.self;
    if(self != null)
        leader.processAck(self.getId(), request.zxid, null);
    else
        LOG.error("Null QuorumPeer");
}