ZOOKEEPER-2789: Reassign ZXID to a 24-bit epoch / 40-bit counter with rolling upgrade

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

@@ -58,6 +58,7 @@
 import org.apache.zookeeper.server.quorum.Leader.PureRequestProposal;
 import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
 import org.apache.zookeeper.server.util.SerializeUtils;
+import org.apache.zookeeper.server.util.ZxidLayoutState;
 import org.apache.zookeeper.server.util.ZxidUtils;
 import org.apache.zookeeper.txn.TxnDigest;
 import org.apache.zookeeper.txn.TxnHeader;
@@ -85,6 +86,13 @@ public class ZKDatabase {
 
     private final boolean allowDiscontinuousProposals = Boolean.getBoolean("zookeeper.test.allowDiscontinuousProposals");
 
+    /** How the zxids in this database are split into epoch and counter; set by the owning QuorumPeer. */
+    private volatile ZxidLayoutState zxidLayoutState = ZxidLayoutState.legacyOnly();
+
+    public void setZxidLayoutState(ZxidLayoutState zxidLayoutState) {
+        this.zxidLayoutState = zxidLayoutState;
+    }
+
     /**
      * Default value is to use snapshot if txnlog size exceeds 1/3 the size of snapshot
      */
@@ -332,7 +340,8 @@ public void addCommittedProposal(Request request) {
                 return;
             } else if (!allowDiscontinuousProposals
                     && request.zxid != maxCommittedLog + 1
-                    && ZxidUtils.getEpochFromZxid(request.zxid) <= ZxidUtils.getEpochFromZxid(maxCommittedLog)) {
+                    && zxidLayoutState.layoutFor(request.zxid).getEpochFromZxid(request.zxid)
+                            <= zxidLayoutState.layoutFor(maxCommittedLog).getEpochFromZxid(maxCommittedLog)) {
                 String msg = String.format(
                     "Committed proposal cached out of order: 0x%s is not the next proposal of 0x%s",
                     ZxidUtils.zxidToString(request.zxid),

zookeeper-server/src/main/java/org/apache/zookeeper/server/admin/Commands.java

@@ -71,6 +71,7 @@
 import org.apache.zookeeper.server.quorum.ReadOnlyZooKeeperServer;
 import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
 import org.apache.zookeeper.server.util.RateLimiter;
+import org.apache.zookeeper.server.util.ZxidLayout;
 import org.apache.zookeeper.server.util.ZxidUtils;
 import org.eclipse.jetty.http.HttpStatus;
 import org.slf4j.Logger;
@@ -1211,8 +1212,9 @@ public CommandResponse runGet(ZooKeeperServer zkServer, Map<String, String> kwar
                 response.put("voting", voting);
                 long lastProcessedZxid = zkServer.getZKDatabase().getDataTreeLastProcessedZxid();
                 response.put("last_zxid", "0x" + ZxidUtils.zxidToString(lastProcessedZxid));
-                response.put("zab_epoch", ZxidUtils.getEpochFromZxid(lastProcessedZxid));
-                response.put("zab_counter", ZxidUtils.getCounterFromZxid(lastProcessedZxid));
+                ZxidLayout layout = peer.getZxidLayoutState().layoutFor(lastProcessedZxid);
+                response.put("zab_epoch", layout.getEpochFromZxid(lastProcessedZxid));
+                response.put("zab_counter", layout.getCounterFromZxid(lastProcessedZxid));
                 response.put("zabstate", zabState.name().toLowerCase());
             } else {
                 response.put("voting", false);

zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/TxnLogToolkit.java

@@ -55,6 +55,7 @@
 import org.apache.zookeeper.server.TxnLogEntry;
 import org.apache.zookeeper.server.util.LogChopper;
 import org.apache.zookeeper.server.util.SerializeUtils;
+import org.apache.zookeeper.server.util.ZxidLayoutState;
 import org.apache.zookeeper.txn.CheckVersionTxn;
 import org.apache.zookeeper.txn.CreateContainerTxn;
 import org.apache.zookeeper.txn.CreateTTLTxn;
@@ -117,6 +118,12 @@ Options getOptions() {
 
     // chop mode
     private long zxid = -1L;
+    // chop mode: the epoch from which the ensemble uses the wide-counter zxid
+    // layout (the content of the zxidLayoutSwitchEpoch file), or -1 when the
+    // log is entirely in the legacy layout. Only affects the gap diagnostics,
+    // which decompose zxids into epoch/counter; the chop boundary itself is a
+    // purely numeric comparison and is layout-independent.
+    private long zxidLayoutSwitchEpoch = -1L;
 
     /**
      * @param args Command line arguments
@@ -164,8 +171,15 @@ public TxnLogToolkit(
     }
 
     public TxnLogToolkit(String txnLogFileName, String zxidName) throws TxnLogToolkitException {
+        this(txnLogFileName, zxidName, null);
+    }
+
+    public TxnLogToolkit(String txnLogFileName, String zxidName, String switchEpochName) throws TxnLogToolkitException {
         txnLogFile = loadTxnFile(txnLogFileName);
         zxid = Long.decode(zxidName);
+        if (switchEpochName != null) {
+            zxidLayoutSwitchEpoch = Long.decode(switchEpochName);
+        }
     }
 
     private File loadTxnFile(String txnLogFileName) throws TxnLogToolkitException {
@@ -253,9 +267,14 @@ public void dump(Scanner scanner) throws Exception {
 
     public void chop() {
         File targetFile = new File(txnLogFile.getParentFile(), txnLogFile.getName() + ".chopped" + zxid);
+        ZxidLayoutState layoutState = ZxidLayoutState.legacyOnly();
+        if (zxidLayoutSwitchEpoch >= 0) {
+            layoutState = new ZxidLayoutState();
+            layoutState.switchAt(zxidLayoutSwitchEpoch);
+        }
         try (InputStream is = new BufferedInputStream(new FileInputStream(txnLogFile));
              OutputStream os = new BufferedOutputStream(new FileOutputStream(targetFile))) {
-            if (!LogChopper.chop(is, os, zxid)) {
+            if (!LogChopper.chop(is, os, zxid, layoutState)) {
                 throw new TxnLogToolkitException(
                     ExitCode.INVALID_INVOCATION.getValue(),
                     "Failed to chop %s",
@@ -457,8 +476,13 @@ private static TxnLogToolkit parseCommandLine(String[] args) throws TxnLogToolki
         // Chop mode options
         Option chopOpt = new Option("c", "chop", false, "Chop mode. Chop txn file to a zxid.");
         Option zxidOpt = new Option("z", "zxid", true, "Used with chop. Zxid to which to chop.");
+        Option switchEpochOpt = new Option("s", "switch-epoch", true,
+            "Used with chop. The wide-counter zxid layout switch epoch (content of the zxidLayoutSwitchEpoch "
+                + "file) when the log spans the layout switch, so gap diagnostics decode zxids correctly. "
+                + "Optional; the chop result is the same with or without it.");
         options.addOption(chopOpt);
         options.addOption(zxidOpt);
+        options.addOption(switchEpochOpt);
 
         try {
             CommandLine cli = parser.parse(options, args);
@@ -469,7 +493,7 @@ private static TxnLogToolkit parseCommandLine(String[] args) throws TxnLogToolki
                 printHelpAndExit(1, options);
             }
             if (cli.hasOption("chop") && cli.hasOption("zxid")) {
-                return new TxnLogToolkit(cli.getArgs()[0], cli.getOptionValue("zxid"));
+                return new TxnLogToolkit(cli.getArgs()[0], cli.getOptionValue("zxid"), cli.getOptionValue("switch-epoch"));
             }
             return new TxnLogToolkit(cli.hasOption("recover"), cli.hasOption("verbose"), cli.getArgs()[0], cli.hasOption("yes"));
         } catch (ParseException e) {
@@ -479,7 +503,7 @@ private static TxnLogToolkit parseCommandLine(String[] args) throws TxnLogToolki
 
     private static void printHelpAndExit(int exitCode, Options options) {
         HelpFormatter help = new HelpFormatter();
-        help.printHelp(120, "TxnLogToolkit [-dhrvc] <txn_log_file_name> (-z <zxid>)", "", options, "");
+        help.printHelp(120, "TxnLogToolkit [-dhrvc] <txn_log_file_name> (-z <zxid>) [-s <switch_epoch>]", "", options, "");
         ServiceUtils.requestSystemExit(exitCode);
     }
 

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

@@ -36,7 +36,6 @@
 import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
 import org.apache.zookeeper.server.quorum.flexible.QuorumOracleMaj;
 import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
-import org.apache.zookeeper.server.util.ZxidUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -284,7 +283,7 @@ public void run() {
                                 }
                             } else {
                                 LOG.info("Backward compatibility mode (28 bits), server id: {}", response.sid);
-                                rpeerepoch = ZxidUtils.getEpochFromZxid(rzxid);
+                                rpeerepoch = self.getZxidLayoutState().layoutFor(rzxid).getEpochFromZxid(rzxid);
                             }
 
                             // check if we have a version that includes config. If so extract config info from message.

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

@@ -93,7 +93,7 @@ void followLeader() throws InterruptedException {
                 }
                 //check to see if the leader zxid is lower than ours
                 //this should never happen but is just a safety check
-                long newEpoch = ZxidUtils.getEpochFromZxid(newEpochZxid);
+                long newEpoch = self.getZxidLayoutState().current().getEpochFromZxid(newEpochZxid);
                 if (newEpoch < self.getAcceptedEpoch()) {
                     LOG.error("Proposed leader epoch "
                               + ZxidUtils.zxidToString(newEpochZxid)

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

@@ -77,7 +77,7 @@ protected void setupRequestProcessors() {
     LinkedBlockingQueue<Request> pendingTxns = new LinkedBlockingQueue<>();
 
     public void logRequest(Request request) {
-        if ((request.zxid & 0xffffffffL) != 0) {
+        if (self.getZxidLayoutState().current().getCounterFromZxid(request.zxid) != 0) {
             pendingTxns.add(request);
         }
         syncProcessor.processRequest(request);

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

@@ -68,7 +68,8 @@
 import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
 import org.apache.zookeeper.server.quorum.auth.QuorumAuthServer;
 import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
-import org.apache.zookeeper.server.util.ZxidUtils;
+import org.apache.zookeeper.server.util.ZxidLayout;
+import org.apache.zookeeper.server.util.ZxidLayoutState;
 import org.apache.zookeeper.util.ServiceUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -654,15 +655,15 @@ void lead() throws IOException, InterruptedException {
 
             long epoch = getEpochToPropose(self.getMyId(), self.getAcceptedEpoch());
 
-            zk.setZxid(ZxidUtils.makeZxid(epoch, 0));
+            zk.setZxid(self.getZxidLayoutState().current().makeZxid(epoch, 0));
 
             synchronized (this) {
                 lastProposed = zk.getZxid();
             }
 
             newLeaderProposal.packet = new QuorumPacket(NEWLEADER, zk.getZxid(), null, null);
 
-            if ((newLeaderProposal.packet.getZxid() & 0xffffffffL) != 0) {
+            if (self.getZxidLayoutState().current().getCounterFromZxid(newLeaderProposal.packet.getZxid()) != 0) {
                 LOG.info("NEWLEADER proposal has Zxid of {}", Long.toHexString(newLeaderProposal.packet.getZxid()));
             }
 
@@ -755,7 +756,7 @@ void lead() throws IOException, InterruptedException {
             String initialZxid = System.getProperty("zookeeper.testingonly.initialZxid");
             if (initialZxid != null) {
                 long zxid = Long.parseLong(initialZxid);
-                zk.setZxid((zk.getZxid() & 0xffffffff00000000L) | zxid);
+                zk.setZxid(self.getZxidLayoutState().current().clearCounter(zk.getZxid()) | zxid);
             }
 
             if (!System.getProperty("zookeeper.leaderServes", "yes").equals("no")) {
@@ -1065,7 +1066,7 @@ public synchronized void processAck(long sid, long zxid, SocketAddress followerA
             LOG.trace("outstanding proposals all");
         }
 
-        if ((zxid & 0xffffffffL) == 0) {
+        if (self.getZxidLayoutState().current().getCounterFromZxid(zxid) == 0) {
             /*
              * We no longer process NEWLEADER ack with this method. However,
              * the learner sends an ack back to the leader after it gets
@@ -1274,7 +1275,15 @@ public synchronized long getLastProposed() {
      * Returns the current epoch of the leader.
      */
     public long getEpoch() {
-        return ZxidUtils.getEpochFromZxid(lastProposed);
+        // Read lastProposed through the synchronized getter: everywhere else
+        // it is accessed under the Leader monitor, so reading it unsynchronized
+        // here would be an inconsistent-synchronization data race.
+        return self.getZxidLayoutState().current().getEpochFromZxid(getLastProposed());
+    }
+
+    @Override
+    public ZxidLayoutState getZxidLayoutState() {
+        return self.getZxidLayoutState();
     }
 
     @SuppressWarnings("serial")
@@ -1298,11 +1307,13 @@ public Proposal propose(Request request) throws XidRolloverException {
             ServiceUtils.requestSystemExit(ExitCode.UNEXPECTED_ERROR.getValue());
         }
         /**
-         * Address the rollover issue. All lower 32bits set indicate a new leader
-         * election. Force a re-election instead. See ZOOKEEPER-1277
+         * Address the rollover issue. An exhausted counter part indicates a
+         * new leader election. Force a re-election instead. See ZOOKEEPER-1277
          */
-        if ((request.zxid & 0xffffffffL) == 0xffffffffL) {
-            String msg = "zxid lower 32 bits have rolled over, forcing re-election, and therefore new epoch start";
+        ZxidLayout layout = self.getZxidLayoutState().current();
+        if (layout.getCounterFromZxid(request.zxid) == layout.getMaxCounter()) {
+            String msg = "zxid counter (lower " + layout.getCounterBits()
+                + " bits) has rolled over, forcing re-election, and therefore new epoch start";
             shutdown(msg);
             throw new XidRolloverException(msg);
         }
@@ -1480,6 +1491,7 @@ public long getEpochToPropose(long sid, long lastAcceptedEpoch) throws Interrupt
             QuorumVerifier verifier = self.getQuorumVerifier();
             if (connectingFollowers.contains(self.getMyId()) && verifier.containsQuorum(connectingFollowers)) {
                 waitingForNewEpoch = false;
+                maybeSwitchZxidLayout(epoch);
                 self.setAcceptedEpoch(epoch);
                 connectingFollowers.notifyAll();
             } else {
@@ -1501,6 +1513,27 @@ public long getEpochToPropose(long sid, long lastAcceptedEpoch) throws Interrupt
         }
     }
 
+    /**
+     * Switches the ensemble to the wide-counter zxid layout at the moment
+     * the new epoch is established, if enabled and not yet switched. Doing
+     * it inside the epoch decision point guarantees that every zxid of the
+     * new epoch — the leader's own as well as those handed to the
+     * LearnerHandlers — is composed with one consistent layout.
+     */
+    private void maybeSwitchZxidLayout(long newEpoch) throws IOException {
+        if (!self.isWideCounterZxidEnabled() || self.getZxidLayoutState().isSwitched()) {
+            return;
+        }
+        if (newEpoch > ZxidLayout.WIDE_COUNTER.getMaxEpoch()) {
+            LOG.warn("Cannot switch to the {} zxid layout: epoch 0x{} exceeds its largest representable epoch 0x{}",
+                ZxidLayout.WIDE_COUNTER,
+                Long.toHexString(newEpoch),
+                Long.toHexString(ZxidLayout.WIDE_COUNTER.getMaxEpoch()));
+            return;
+        }
+        self.recordZxidLayoutSwitch(newEpoch);
+    }
+
     @Override
     public ZKDatabase getZKDatabase() {
         return zk.getZKDatabase();