From 44bb60f739507dff2b9f9a8b2d4b13f09e383f7a Mon Sep 17 00:00:00 2001 From: Jeffrey Xiao Date: Fri, 30 Sep 2022 10:43:53 -0400 Subject: [PATCH 1/2] Add failing test demonstrating ZOOKEEPER-4394. This test demonstrates that a leader sending outstanding proposals with a DIFF will cause an NPE if the commit to that proposal comes before the UPTODATE. --- .../zookeeper/server/quorum/LearnerTest.java | 83 +++++++++++++++++++ 1 file changed, 83 insertions(+) diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LearnerTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LearnerTest.java index d876c167f94..e3b019e5672 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LearnerTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LearnerTest.java @@ -364,4 +364,87 @@ public void accept(Integer exitCode) { TestUtils.deleteFileRecursively(tmpFile); } } + + /* We need this to be a `FollowerZooKeeperServer` instead of a + * `LearnerZooKeeperServer` because otherwise it would throw an "Unknown + * server type" exception when finishing the sync from the leader. + */ + static class SimpleFollowerZooKeeperServer extends FollowerZooKeeperServer { + + Learner learner; + + public SimpleFollowerZooKeeperServer(FileTxnSnapLog ftsl, QuorumPeer self) + throws IOException { + super(ftsl, self, new ZKDatabase(ftsl)); + } + + @Override + public Learner getLearner() { + return learner; + } + } + + static class SimpleLearnerWithFollower extends Learner { + SimpleLearnerWithFollower(FileTxnSnapLog ftsl) throws IOException { + self = new QuorumPeer(); + self.setTxnFactory(ftsl); + zk = new SimpleFollowerZooKeeperServer(ftsl, self); + ((SimpleFollowerZooKeeperServer) zk).learner = this; + } + } + + + @Test + public void diffWithOutstandingProposalsDoesNotNPE() throws Exception { + File tmpDir = File.createTempFile("test", ".dir", testData); + tmpDir.delete(); + try { + FileTxnSnapLog ftsl = new FileTxnSnapLog(tmpDir, tmpDir); + SimpleLearnerWithFollower sl = new SimpleLearnerWithFollower(ftsl); + + // Set up bogus streams + sl.leaderOs = BinaryOutputArchive.getArchive(new ByteArrayOutputStream()); + sl.sock = new Socket(); + sl.bufferedOutput = new BufferedOutputStream(new ByteArrayOutputStream()); + + // fake messages from server + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + BinaryOutputArchive oa = BinaryOutputArchive.getArchive(baos); + oa.writeRecord(new QuorumPacket(Leader.DIFF, 1, null, null), null); + TxnHeader hdr = new TxnHeader(0, 1, 1, 0, ZooDefs.OpCode.create); + CreateTxn txn = new CreateTxn("/foo", new byte[0], new ArrayList(), false, sl.zk.getZKDatabase().getNode("/").stat.getCversion()); + ByteArrayOutputStream tbaos = new ByteArrayOutputStream(); + BinaryOutputArchive boa = BinaryOutputArchive.getArchive(tbaos); + hdr.serialize(boa, "hdr"); + txn.serialize(boa, "txn"); + tbaos.close(); + oa.writeRecord(new QuorumPacket(Leader.PROPOSAL, 1, tbaos.toByteArray(), null), null); + oa.writeRecord(new QuorumPacket(Leader.COMMIT, 1, null, null), null); + + hdr = new TxnHeader(0, 2, 2, 0, ZooDefs.OpCode.create); + txn = new CreateTxn("/bar", new byte[0], new ArrayList(), false, sl.zk.getZKDatabase().getNode("/").stat.getCversion()); + tbaos = new ByteArrayOutputStream(); + boa = BinaryOutputArchive.getArchive(tbaos); + hdr.serialize(boa, "hdr"); + txn.serialize(boa, "txn"); + tbaos.close(); + oa.writeRecord(new QuorumPacket(Leader.PROPOSAL, 2, tbaos.toByteArray(), null), null); + oa.writeRecord(new QuorumPacket(Leader.NEWLEADER, 1, null, null), null); + oa.writeRecord(new QuorumPacket(Leader.COMMIT, 2, null, null), null); + oa.writeRecord(new QuorumPacket(Leader.UPTODATE, -1, null, null), null); + baos.close(); + + // setup the messages to be streamed to follower + sl.leaderIs = BinaryInputArchive.getArchive(new ByteArrayInputStream(baos.toByteArray())); + + try { + sl.syncWithLeader(1); + } catch (EOFException e) {} + + sl.zk.shutdown(); + assertEquals(2, sl.zk.getLastProcessedZxid()); + } finally { + TestUtils.deleteFileRecursively(tmpDir); + } + } } From 22a0e46ec4018fc3feacc645ef546b36a40f031e Mon Sep 17 00:00:00 2001 From: Jeffrey Xiao Date: Fri, 30 Sep 2022 10:44:31 -0400 Subject: [PATCH 2/2] ZOOKEEPER-4394: Fix NPE when receiving diff with outstanding proposal This commit fixes the failing test and addresses ZOOKEEPER-4394. The main problem with the learner code is that the purpose of `packetsNotCommitted` is a bit overloaded -- it represents two things: 1. The packets that we need to durably log. 2. The packets that we haven't committed yet. However, there are cases where these two sets of packets are not identical. For example, suppose we are catching up via DIFF and we receive an outstanding PROPOSAL before NEWLEADER and the COMMIT after NEWLEADER, but before UPTODATE. In this case, we need to durably log all proposals before we ACK the NEWLEADER, but we still need to apply the PROPOSAL when we receive the commit. This commit separates `packetsNotCommitted` into an additional field `packetsNotLogged` that represents the first set of packets. --- .../zookeeper/server/quorum/Learner.java | 49 ++++++++++--------- 1 file changed, 27 insertions(+), 22 deletions(-) diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Learner.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Learner.java index b6eeb758ac9..dad36784622 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Learner.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Learner.java @@ -555,6 +555,7 @@ protected void syncWithLeader(long newLeaderZxid) throws Exception { boolean syncSnapshot = false; readPacket(qp); Deque packetsCommitted = new ArrayDeque<>(); + Deque packetsNotLogged = new ArrayDeque<>(); Deque packetsNotCommitted = new ArrayDeque<>(); synchronized (zk) { if (qp.getType() == Leader.DIFF) { @@ -643,33 +644,37 @@ protected void syncWithLeader(long newLeaderZxid) throws Exception { self.setLastSeenQuorumVerifier(qv, true); } + packetsNotLogged.add(pif); packetsNotCommitted.add(pif); break; case Leader.COMMIT: case Leader.COMMITANDACTIVATE: pif = packetsNotCommitted.peekFirst(); - if (pif.hdr.getZxid() == qp.getZxid() && qp.getType() == Leader.COMMITANDACTIVATE) { - QuorumVerifier qv = self.configFromString(new String(((SetDataTxn) pif.rec).getData(), UTF_8)); - boolean majorChange = self.processReconfig( - qv, - ByteBuffer.wrap(qp.getData()).getLong(), qp.getZxid(), - true); - if (majorChange) { + if (pif.hdr.getZxid() != qp.getZxid()) { + LOG.warn( + "Committing 0x{}, but next proposal is 0x{}", + Long.toHexString(qp.getZxid()), + Long.toHexString(pif.hdr.getZxid())); + } else { + if (qp.getType() == Leader.COMMITANDACTIVATE) { + QuorumVerifier qv = self.configFromString(new String(((SetDataTxn) pif.rec).getData(), UTF_8)); + boolean majorChange = self.processReconfig( + qv, + ByteBuffer.wrap(qp.getData()).getLong(), qp.getZxid(), + true); + if (majorChange) { throw new Exception("changes proposed in reconfig"); + } } - } - if (!writeToTxnLog) { - if (pif.hdr.getZxid() != qp.getZxid()) { - LOG.warn( - "Committing 0x{}, but next proposal is 0x{}", - Long.toHexString(qp.getZxid()), - Long.toHexString(pif.hdr.getZxid())); - } else { + if (!writeToTxnLog) { + // Apply to db directly if we haven't taken the snapshot. zk.processTxn(pif.hdr, pif.rec); + packetsNotLogged.remove(); + packetsNotCommitted.remove(); + } else { packetsNotCommitted.remove(); + packetsCommitted.add(qp.getZxid()); } - } else { - packetsCommitted.add(qp.getZxid()); } break; case Leader.INFORM: @@ -708,7 +713,7 @@ protected void syncWithLeader(long newLeaderZxid) throws Exception { // Apply to db directly if we haven't taken the snapshot zk.processTxn(packet.hdr, packet.rec); } else { - packetsNotCommitted.add(packet); + packetsNotLogged.add(packet); packetsCommitted.add(qp.getZxid()); } @@ -756,10 +761,10 @@ protected void syncWithLeader(long newLeaderZxid) throws Exception { zk.startupWithoutServing(); if (zk instanceof FollowerZooKeeperServer) { FollowerZooKeeperServer fzk = (FollowerZooKeeperServer) zk; - for (PacketInFlight p : packetsNotCommitted) { + for (PacketInFlight p : packetsNotLogged) { fzk.logRequest(p.hdr, p.rec, p.digest); } - packetsNotCommitted.clear(); + packetsNotLogged.clear(); } writePacket(new QuorumPacket(Leader.ACK, newLeaderZxid, null, null), true); @@ -782,7 +787,7 @@ protected void syncWithLeader(long newLeaderZxid) throws Exception { // We need to log the stuff that came in between the snapshot and the uptodate if (zk instanceof FollowerZooKeeperServer) { FollowerZooKeeperServer fzk = (FollowerZooKeeperServer) zk; - for (PacketInFlight p : packetsNotCommitted) { + for (PacketInFlight p : packetsNotLogged) { fzk.logRequest(p.hdr, p.rec, p.digest); } for (Long zxid : packetsCommitted) { @@ -792,7 +797,7 @@ protected void syncWithLeader(long newLeaderZxid) throws Exception { // Similar to follower, we need to log requests between the snapshot // and UPTODATE ObserverZooKeeperServer ozk = (ObserverZooKeeperServer) zk; - for (PacketInFlight p : packetsNotCommitted) { + for (PacketInFlight p : packetsNotLogged) { Long zxid = packetsCommitted.peekFirst(); if (p.hdr.getZxid() != zxid) { // log warning message if there is no matching commit