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 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); + } + } }