y

We wish to fix this long-standing issue in the code.
Note that the previous commit processor algorithm had the same approach (as the one suggested in this fix) when dealing with a request that has a different cxid than session's expected one (see [here](https://github.com/apache/zookeeper/commit/9fc632c4f0a340b0a00ec6dff39c7b454c802822#diff-5cc688a027068714af01b0ad4d292fe5L238)).

This fix is based on the code from https://github.com/apache/zookeeper/pull/167, following the discussion in https://issues.apache.org/jira/browse/ZOOKEEPER-2684 .

Author: Kfir Lev-Ari <klevari@apple.com>

Reviewers: Alexander Shraer <shralex@apache.org>, Abraham Fine <afine@apache.org>

Closes #411 from kfirlevari/ZOOKEEPER-2684

src/java/main/org/apache/zookeeper/server/quorum/CommitProcessor.java

@@ -246,8 +246,7 @@ public class CommitProcessor extends ZooKeeperCriticalThread implements
                     }
 
                     /*
-                     * Check if request is pending, if so, update it with the
-                     * committed info
+                     * Check if request is pending, if so, update it with the committed info
                      */
                     LinkedList<Request> sessionQueue = pendingRequests
                             .get(request.sessionId);
@@ -255,24 +254,52 @@ public class CommitProcessor extends ZooKeeperCriticalThread implements
                         // If session queue != null, then it is also not empty.
                         Request topPending = sessionQueue.poll();
                         if (request.cxid != topPending.cxid) {
-                            LOG.error(
-                                    "Got cxid 0x"
-                                            + Long.toHexString(request.cxid)
-                                            + " expected 0x" + Long.toHexString(
-                                                    topPending.cxid)
-                                    + " for client session id "
-                                    + Long.toHexString(request.sessionId));
-                            throw new IOException("Error: unexpected cxid for"
-                                    + "client session");
+                            /*
+                             * TL;DR - we should not encounter this scenario often under normal load.
+                             * We pass the commit to the next processor and put the pending back with a warning.
+                             *
+                             * Generally, we can get commit requests that are not at the queue head after
+                             * a session moved (see ZOOKEEPER-2684). Let's denote the previous server of the session
+                             * with A, and the server that the session moved to with B (keep in mind that it is
+                             * possible that the session already moved from B to a new server C, and maybe C=A).
+                             * 1. If request.cxid < topPending.cxid : this means that the session requested this update
+                             * from A, then moved to B (i.e., which is us), and now B receives the commit
+                             * for the update after the session already performed several operations in B
+                             * (and therefore its cxid is higher than that old request).
+                             * 2. If request.cxid > topPending.cxid : this means that the session requested an updated
+                             * from B with cxid that is bigger than the one we know therefore in this case we
+                             * are A, and we lost the connection to the session. Given that we are waiting for a commit
+                             * for that update, it means that we already sent the request to the leader and it will
+                             * be committed at some point (in this case the order of cxid won't follow zxid, since zxid
+                             * is an increasing order). It is not safe for us to delete the session's queue at this
+                             * point, since it is possible that the session has newer requests in it after it moved
+                             * back to us. We just leave the queue as it is, and once the commit arrives (for the old
+                             * request), the finalRequestProcessor will see a closed cnxn handle, and just won't send a
+                             * response.
+                             * Also note that we don't have a local session, therefore we treat the request
+                             * like any other commit for a remote request, i.e., we perform the update without sending
+                             * a response.
+                             */
+                            LOG.warn("Got request " + request +
+                                    " but we are expecting request " + topPending);
+                            sessionQueue.addFirst(topPending);
+                        } else {
+                            /*
+                             * Generally, we want to send to the next processor our version of the request,
+                             * since it contains the session information that is needed for post update processing.
+                             * In more details, when a request is in the local queue, there is (or could be) a client
+                             * attached to this server waiting for a response, and there is other bookkeeping of
+                             * requests that are outstanding and have originated from this server
+                             * (e.g., for setting the max outstanding requests) - we need to update this info when an
+                             * outstanding request completes. Note that in the other case (above), the operation
+                             * originated from a different server and there is no local bookkeeping or a local client
+                             * session that needs to be notified.
+                             */
+                            topPending.setHdr(request.getHdr());
+                            topPending.setTxn(request.getTxn());
+                            topPending.zxid = request.zxid;
+                            request = topPending;
                         }
-                        /*
-                         * We want to send our version of the request. the
-                         * pointer to the connection in the request
-                         */
-                        topPending.setHdr(request.getHdr());
-                        topPending.setTxn(request.getTxn());
-                        topPending.zxid = request.zxid;
-                        request = topPending;
                     }
 
                     sendToNextProcessor(request);

src/java/test/org/apache/zookeeper/server/quorum/CommitProcessorConcurrencyTest.java

@@ -108,6 +108,7 @@ public class CommitProcessorConcurrencyTest extends ZKTestCase {
 
                 @Override
                 public void notifyStopping(String threadName, int errorCode) {
+                    Assert.fail("Commit processor crashed " + errorCode);
                 }
             });
         }
@@ -376,4 +377,129 @@ public class CommitProcessorConcurrencyTest extends ZKTestCase {
                     !processedRequests.contains(r));
         }
     }
+
+    /**
+     * In the following test, we verify that we can handle the case that we got a commit
+     * of a request we never seen since the session that we just established. This can happen
+     * when a session is just established and there is request waiting to be committed in the
+     * session queue but it sees a commit for a request that belongs to the previous connection.
+     */
+    @Test(timeout = 5000)
+    public void noCrashOnCommittedRequestsOfUnseenRequestTest() throws Exception {
+        final String path = "/noCrash/OnCommittedRequests/OfUnseenRequestTest";
+        final int numberofReads = 10;
+        final int sessionid = 0x123456;
+        final int firstCXid = 0x100;
+        int readReqId = firstCXid;
+        processor.stoppedMainLoop = true;
+        HashSet<Request> localRequests = new HashSet<Request>();
+        // queue the blocking write request to queuedRequests
+        Request firstCommittedReq = newRequest(
+                new CreateRequest(path, new byte[0], Ids.OPEN_ACL_UNSAFE,
+                        CreateMode.PERSISTENT_SEQUENTIAL.toFlag()),
+                OpCode.create, sessionid, readReqId++);
+        processor.queuedRequests.add(firstCommittedReq);
+        localRequests.add(firstCommittedReq);
+
+        // queue read requests to queuedRequests
+        for (; readReqId <= numberofReads+firstCXid; ++readReqId) {
+            Request readReq = newRequest(new GetDataRequest(path, false),
+                    OpCode.getData, sessionid, readReqId);
+            processor.queuedRequests.add(readReq);
+            localRequests.add(readReq);
+        }
+
+        //run once
+        Assert.assertTrue(processor.queuedRequests.containsAll(localRequests));
+        processor.initThreads(defaultSizeOfThreadPool);
+        processor.run();
+        Thread.sleep(1000);
+
+        //We verify that the processor is waiting for the commit
+        Assert.assertTrue(processedRequests.isEmpty());
+
+        // We add a commit that belongs to the same session but with smaller cxid,
+        // i.e., commit of an update from previous connection of this session.
+        Request preSessionCommittedReq = newRequest(
+                new CreateRequest(path, new byte[0], Ids.OPEN_ACL_UNSAFE,
+                        CreateMode.PERSISTENT_SEQUENTIAL.toFlag()),
+                OpCode.create, sessionid, firstCXid - 2);
+        processor.committedRequests.add(preSessionCommittedReq);
+        processor.committedRequests.add(firstCommittedReq);
+        processor.run();
+        Thread.sleep(1000);
+
+        //We verify that the commit processor processed the old commit prior to the newer messages
+        Assert.assertTrue(processedRequests.peek() == preSessionCommittedReq);
+
+        processor.run();
+        Thread.sleep(1000);
+
+        //We verify that the commit processor handle all messages.
+        Assert.assertTrue(processedRequests.containsAll(localRequests));
+    }
+
+    /**
+     * In the following test, we verify if we handle the case in which we get a commit
+     * for a request that has higher Cxid than the one we are waiting. This can happen
+     * when a session connection is lost but there is a request waiting to be committed in the
+     * session queue. However, since the session has moved, new requests can get to
+     * the leader out of order. Hence, the commits can also arrive "out of order" w.r.t. cxid.
+     * We should commit the requests according to the order we receive from the leader, i.e., wait for the relevant commit.
+     */
+    @Test(timeout = 5000)
+    public void noCrashOnOutofOrderCommittedRequestTest() throws Exception {
+        final String path = "/noCrash/OnCommittedRequests/OfUnSeenRequestTest";
+        final int sessionid = 0x123456;
+        final int lastCXid = 0x100;
+        final int numberofReads = 10;
+        int readReqId = lastCXid;
+        processor.stoppedMainLoop = true;
+        HashSet<Request> localRequests = new HashSet<Request>();
+
+        // queue the blocking write request to queuedRequests
+        Request orphanCommittedReq = newRequest(
+                new CreateRequest(path, new byte[0], Ids.OPEN_ACL_UNSAFE,
+                        CreateMode.PERSISTENT_SEQUENTIAL.toFlag()),
+                OpCode.create, sessionid, lastCXid);
+        processor.queuedRequests.add(orphanCommittedReq);
+        localRequests.add(orphanCommittedReq);
+
+        // queue read requests to queuedRequests
+        for (; readReqId <= numberofReads+lastCXid; ++readReqId) {
+            Request readReq = newRequest(new GetDataRequest(path, false),
+                    OpCode.getData, sessionid, readReqId);
+            processor.queuedRequests.add(readReq);
+            localRequests.add(readReq);
+        }
+
+        //run once
+        processor.initThreads(defaultSizeOfThreadPool);
+        processor.run();
+        Thread.sleep(1000);
+
+        //We verify that the processor is waiting for the commit
+        Assert.assertTrue(processedRequests.isEmpty());
+
+        // We add a commit that belongs to the same session but with larger cxid,
+        // i.e., commit of an update from the next connection of this session.
+        Request otherSessionCommittedReq = newRequest(
+                new CreateRequest(path, new byte[0], Ids.OPEN_ACL_UNSAFE,
+                        CreateMode.PERSISTENT_SEQUENTIAL.toFlag()),
+                OpCode.create, sessionid, lastCXid+10);
+        processor.committedRequests.add(otherSessionCommittedReq);
+        processor.committedRequests.add(orphanCommittedReq);
+        processor.run();
+        Thread.sleep(1000);
+
+        //We verify that the commit processor processed the old commit prior to the newer messages
+        Assert.assertTrue(processedRequests.size() == 1);
+        Assert.assertTrue(processedRequests.contains(otherSessionCommittedReq));
+
+        processor.run();
+        Thread.sleep(1000);
+
+        //We verify that the commit processor handle all messages.
+        Assert.assertTrue(processedRequests.containsAll(localRequests));
+    }
 }