ZOOKEEPER-3594: Ability to skip proposing requests with error transactions

zookeeper-server/src/main/java/org/apache/zookeeper/server/ExitCode.java

@@ -48,7 +48,10 @@ public enum ExitCode {
     QUORUM_PACKET_ERROR(13),
 
     /** Unable to bind to the quorum (election) port after multiple retry */
-    UNABLE_TO_BIND_QUORUM_PORT(14);
+    UNABLE_TO_BIND_QUORUM_PORT(14),
+
+    /** Used to detect if skip txn ran onto the ordering issue */
+    QUORUM_PEER_SKIP_OUT_OF_ORDER(15);
 
     private final int value;
 

zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java

@@ -844,8 +844,22 @@ protected void pRequest(Request request) throws RequestProcessorException {
                 request.getHdr().setType(OpCode.error);
                 request.setTxn(new ErrorTxn(Code.MARSHALLINGERROR.intValue()));
             }
+        } finally {
+            // When skipping error requests is enabled then we want to use next zxid only when we
+            // have a valid request that passed all the validations in this request processor
+            // otherwise we will skip the current request
+            if (LeaderZooKeeperServer.isSkipTxnEnabled() && request.hasError() && request.canSkip()) {
+                request.setSkipped();
+            } else {
+                // When the skip feature is disabled, we will issue a new zxid for every write
+                // request that goes through this request processor and passes the validations
+                if (request.getHdr() != null) {
+                    zks.incrementZxid();
+                }
+            }
         }
         request.zxid = zks.getZxid();
+        request.getHdr().setZxid(zks.getZxid());
         ServerMetrics.getMetrics().PREP_PROCESS_TIME.add(Time.currentElapsedTime() - request.prepStartTime);
         nextProcessor.processRequest(request);
     }

zookeeper-server/src/main/java/org/apache/zookeeper/server/Request.java

@@ -27,6 +27,8 @@
 import org.apache.zookeeper.data.Id;
 import org.apache.zookeeper.metrics.Summary;
 import org.apache.zookeeper.metrics.SummarySet;
+import org.apache.zookeeper.server.quorum.LearnerHandler;
+import org.apache.zookeeper.server.quorum.SkipRequestHandler;
 import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
 import org.apache.zookeeper.server.util.AuthUtil;
 import org.apache.zookeeper.txn.TxnHeader;
@@ -48,6 +50,8 @@ public class Request {
     // associated session timeout. Disabled by default.
     private static volatile boolean staleLatencyCheck = Boolean.parseBoolean(System.getProperty("zookeeper.request_stale_latency_check", "false"));
 
+    private SkipRequestHandler skipRequestHandler;
+
     public Request(ServerCnxn cnxn, long sessionId, int xid, int type, ByteBuffer bb, List<Id> authInfo) {
         this.cnxn = cnxn;
         this.sessionId = sessionId;
@@ -87,6 +91,8 @@ public Request(long sessionId, int xid, int type, TxnHeader hdr, Record txn, lon
 
     public final List<Id> authInfo;
 
+    private boolean isSkipped = false;
+
     public final long createTime = Time.currentElapsedTime();
 
     public long prepQueueStartTime = -1;
@@ -105,6 +111,14 @@ public Request(long sessionId, int xid, int type, TxnHeader hdr, Record txn, lon
 
     public QuorumVerifier qv = null;
 
+    public boolean isFromLeader() {
+        return owner == ServerCnxn.me;
+    }
+
+    public boolean isFromFollower() {
+        return owner instanceof LearnerHandler;
+    }
+
     /**
      * If this is a create or close request for a local-only session.
      */
@@ -462,4 +476,48 @@ public String getUsers() {
         }
         return users.toString();
     }
+
+    /**
+     * This will check if the request's handler supports skip feature,
+     * the information that is passed from Learner to LearnerMaster using protocol version info
+     *
+     * See org.apache.zookeeper.server.quorum.QuorumZooKeeperServer.PROTOCOL_VERSION
+     *
+     */
+    public boolean canSkip() {
+        return skipRequestHandler != null;
+    }
+
+    /**
+     * Will mark this request for request skipping
+     */
+    public void setSkipped() {
+        isSkipped = true;
+    }
+
+    /**
+     * Once we mark request as skipped this method will be returning true
+     */
+    public boolean isSkipped() {
+        return isSkipped;
+    }
+
+    /**
+     * Each request has access to its handler so know who is this requests owner
+     * and interact with it when we need to
+     */
+    public void setSkipRequestHandler(SkipRequestHandler skipRequestHandler) {
+        this.skipRequestHandler = skipRequestHandler;
+    }
+
+    public SkipRequestHandler getSkipRequestHandler() {
+        return skipRequestHandler;
+    }
+
+    /**
+     * Check the result of the PrepRequestProcessor validations for write requests.
+     */
+    public boolean hasError() {
+        return hdr != null && e != null;
+    }
 }

zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerMetrics.java

@@ -151,6 +151,9 @@ private ServerMetrics(MetricsProvider metricsProvider) {
         WRITES_QUEUED_IN_COMMIT_PROCESSOR = metricsContext.getSummary("write_commit_proc_req_queued", DetailLevel.BASIC);
         COMMITS_QUEUED_IN_COMMIT_PROCESSOR = metricsContext.getSummary("commit_commit_proc_req_queued", DetailLevel.BASIC);
         COMMITS_QUEUED = metricsContext.getCounter("request_commit_queued");
+        QUORUM_PEER_SKIP_SENT = metricsContext.getCounter("quorum_peer_skip_sent");
+        QUORUM_PEER_SKIP_OUT_OF_ORDER = metricsContext.getCounter("quorum_peer_skip_out_of_order");
+        QUORUM_PEER_SKIP_QUEUE_SIZE = metricsContext.getCounter("quorum_peer_skip_queue_size");
         READS_ISSUED_IN_COMMIT_PROC = metricsContext.getSummary("read_commit_proc_issued", DetailLevel.BASIC);
         WRITES_ISSUED_IN_COMMIT_PROC = metricsContext.getSummary("write_commit_proc_issued", DetailLevel.BASIC);
 
@@ -183,6 +186,7 @@ private ServerMetrics(MetricsProvider metricsProvider) {
          */
         OM_PROPOSAL_PROCESS_TIME = metricsContext.getSummary("om_proposal_process_time_ms", DetailLevel.ADVANCED);
         OM_COMMIT_PROCESS_TIME = metricsContext.getSummary("om_commit_process_time_ms", DetailLevel.ADVANCED);
+        OM_SKIP_PROCESS_TIME = metricsContext.getSummary("om_skip_process_time_ms", DetailLevel.ADVANCED);
 
         /**
          * Time spent by the final processor. This is tracked in the commit processor.
@@ -193,8 +197,10 @@ private ServerMetrics(MetricsProvider metricsProvider) {
         PROPOSAL_LATENCY = metricsContext.getSummary("proposal_latency", DetailLevel.ADVANCED);
         PROPOSAL_ACK_CREATION_LATENCY = metricsContext.getSummary("proposal_ack_creation_latency", DetailLevel.ADVANCED);
         COMMIT_PROPAGATION_LATENCY = metricsContext.getSummary("commit_propagation_latency", DetailLevel.ADVANCED);
+        SKIP_PROPAGATION_LATENCY = metricsContext.getSummary("skip_propagation_latency", DetailLevel.ADVANCED);
         LEARNER_PROPOSAL_RECEIVED_COUNT = metricsContext.getCounter("learner_proposal_received_count");
         LEARNER_COMMIT_RECEIVED_COUNT = metricsContext.getCounter("learner_commit_received_count");
+        LEARNER_SKIP_RECEIVED_COUNT = metricsContext.getCounter("learner_skip_received_count");
 
         /**
          * Learner handler quorum packet metrics.
@@ -218,6 +224,7 @@ private ServerMetrics(MetricsProvider metricsProvider) {
         QUORUM_ACK_LATENCY = metricsContext.getSummary("quorum_ack_latency", DetailLevel.ADVANCED);
         ACK_LATENCY = metricsContext.getSummarySet("ack_latency", DetailLevel.ADVANCED);
         PROPOSAL_COUNT = metricsContext.getCounter("proposal_count");
+        SKIP_COUNT = metricsContext.getCounter("skip_count");
         QUIT_LEADING_DUE_TO_DISLOYAL_VOTER = metricsContext.getCounter("quit_leading_due_to_disloyal_voter");
 
         STALE_REQUESTS = metricsContext.getCounter("stale_requests");
@@ -304,8 +311,10 @@ private ServerMetrics(MetricsProvider metricsProvider) {
     public final Summary PROPOSAL_LATENCY;
     public final Summary PROPOSAL_ACK_CREATION_LATENCY;
     public final Summary COMMIT_PROPAGATION_LATENCY;
+    public final Summary SKIP_PROPAGATION_LATENCY;
     public final Counter LEARNER_PROPOSAL_RECEIVED_COUNT;
     public final Counter LEARNER_COMMIT_RECEIVED_COUNT;
+    public final Counter LEARNER_SKIP_RECEIVED_COUNT;
 
     public final Summary STARTUP_TXNS_LOADED;
     public final Summary STARTUP_TXNS_LOAD_TIME;
@@ -323,6 +332,7 @@ private ServerMetrics(MetricsProvider metricsProvider) {
     public final Summary QUORUM_ACK_LATENCY;
     public final SummarySet ACK_LATENCY;
     public final Counter PROPOSAL_COUNT;
+    public final Counter SKIP_COUNT;
     public final Counter QUIT_LEADING_DUE_TO_DISLOYAL_VOTER;
 
     /**
@@ -376,6 +386,9 @@ private ServerMetrics(MetricsProvider metricsProvider) {
     public final Summary WRITES_QUEUED_IN_COMMIT_PROCESSOR;
     public final Summary COMMITS_QUEUED_IN_COMMIT_PROCESSOR;
     public final Counter COMMITS_QUEUED;
+    public final Counter QUORUM_PEER_SKIP_SENT;
+    public final Counter QUORUM_PEER_SKIP_OUT_OF_ORDER;
+    public final Counter QUORUM_PEER_SKIP_QUEUE_SIZE;
     public final Summary READS_ISSUED_IN_COMMIT_PROC;
     public final Summary WRITES_ISSUED_IN_COMMIT_PROC;
 
@@ -408,6 +421,7 @@ private ServerMetrics(MetricsProvider metricsProvider) {
      */
     public final Summary OM_PROPOSAL_PROCESS_TIME;
     public final Summary OM_COMMIT_PROCESS_TIME;
+    public final Summary OM_SKIP_PROCESS_TIME;
 
     /**
      * Time spent by the final processor. This is tracked in the commit processor.

zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServer.java

@@ -561,7 +561,11 @@ public SessionTracker getSessionTracker() {
     }
 
     long getNextZxid() {
-        return hzxid.incrementAndGet();
+        return hzxid.get() + 1;
+    }
+
+    void incrementZxid() {
+        hzxid.incrementAndGet();
     }
 
     public void setZxid(long zxid) {
@@ -1716,7 +1720,7 @@ public ProcessTxnResult processTxn(Request request) {
             }
 
             // do not add non quorum packets to the queue.
-            if (quorumRequest) {
+            if (quorumRequest && !request.isSkipped()) {
                 getZKDatabase().addCommittedProposal(request);
             }
             return rc;

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Follower.java

@@ -195,6 +195,18 @@ protected void processPacket(QuorumPacket qp) throws Exception {
                 ServerMetrics.getMetrics().OM_PROPOSAL_PROCESS_TIME.add(Time.currentElapsedTime() - startTime);
             }
             break;
+        case Leader.SKIP:
+            ServerMetrics.getMetrics().LEARNER_SKIP_RECEIVED_COUNT.add(1);
+            fzk.skip(qp.getData());
+            if (om != null) {
+                final long startTime = Time.currentElapsedTime();
+                // We need to create a new QuorumPacket object that will be queued for sending
+                QuorumPacket observerPacket = new QuorumPacket(
+                        qp.getType(), qp.getZxid(), qp.getData(), qp.getAuthinfo());
+                om.proposalSkipped(observerPacket);
+                ServerMetrics.getMetrics().OM_SKIP_PROCESS_TIME.add(Time.currentElapsedTime() - startTime);
+            }
+            break;
         case Leader.COMMIT:
             ServerMetrics.getMetrics().LEARNER_COMMIT_RECEIVED_COUNT.add(1);
             fzk.commit(qp.getZxid());

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/FollowerZooKeeperServer.java

@@ -31,8 +31,10 @@
 import org.apache.zookeeper.server.RequestProcessor;
 import org.apache.zookeeper.server.ServerMetrics;
 import org.apache.zookeeper.server.SyncRequestProcessor;
+import org.apache.zookeeper.server.TxnLogEntry;
 import org.apache.zookeeper.server.ZKDatabase;
 import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
+import org.apache.zookeeper.server.util.SerializeUtils;
 import org.apache.zookeeper.txn.TxnHeader;
 import org.apache.zookeeper.util.ServiceUtils;
 import org.slf4j.Logger;
@@ -108,6 +110,19 @@ public void commit(long zxid) {
         commitProcessor.commit(request);
     }
 
+
+    public void skip(byte[] data) {
+        try {
+            TxnHeader hdr = new TxnHeader();
+            Record txn = SerializeUtils.deserializeTxn(data, hdr);
+            Request request = new Request(hdr.getClientId(), hdr.getCxid(), hdr.getType(), hdr, txn, 0);
+            request.logLatency(ServerMetrics.getMetrics().SKIP_PROPAGATION_LATENCY);
+            commitProcessor.commit(request);
+        } catch (IOException e) {
+            LOG.error("Could not deserialize SKIP request");
+        }
+    }
+
     public synchronized void sync() {
         if (pendingSyncs.size() == 0) {
             LOG.warn("Not expecting a sync.");

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Leader.java

@@ -423,6 +423,13 @@ Optional<ServerSocket> createServerSocket(InetSocketAddress address, boolean por
      */
     static final int INFORMANDACTIVATE = 19;
 
+    /**
+     * This message type tells the Learner that the request has
+     * an error transaction in it which means that it wasn't preceded
+     * with a PROPOSAL and that it can be committed immediately.
+     */
+    final static int SKIP = 101;
+
     final ConcurrentMap<Long, Proposal> outstandingProposals = new ConcurrentHashMap<Long, Proposal>();
 
     private final ConcurrentLinkedQueue<Proposal> toBeApplied = new ConcurrentLinkedQueue<Proposal>();
@@ -955,7 +962,7 @@ public synchronized boolean tryToCommit(Proposal p, long zxid, SocketAddress fol
             commit(zxid);
             inform(p);
         }
-        zk.commitProcessor.commit(p.request);
+        zk.getLeaderHandler().processCommit(p.request);
         if (pendingSyncs.containsKey(zxid)) {
             for (LearnerSyncRequest r : pendingSyncs.remove(zxid)) {
                 sendSync(r);
@@ -1084,6 +1091,9 @@ public void processRequest(Request request) throws RequestProcessorException {
             // request.zxid here because that is set on read requests to equal
             // the zxid of the last write op.
             if (request.getHdr() != null) {
+                if (request.isSkipped()) {
+                    return;
+                }
                 long zxid = request.getHdr().getZxid();
                 Iterator<Proposal> iter = leader.toBeApplied.iterator();
                 if (iter.hasNext()) {
@@ -1613,6 +1623,8 @@ public static String getPacketType(int packetType) {
             return "PROPOSAL";
         case ACK:
             return "ACK";
+        case SKIP:
+            return "SKIP";
         case COMMIT:
             return "COMMIT";
         case COMMITANDACTIVATE: