ZOOKEEPER-4643, ZOOKEEPER-4646 & ZOOKEEPER-4394: Committed txns lost / NullPointerException in Learner.syncWithLeader() #2152
+199
−56
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -741,8 +741,8 @@ public void converseWithFollower(InputArchive ia, OutputArchive oa, Follower f) | |
|
|
||
| readPacketSkippingPing(ia, qp); | ||
| assertEquals(Leader.ACKEPOCH, qp.getType()); | ||
| assertEquals(ZxidUtils.makeZxid(0, 0), qp.getZxid()); | ||
| assertEquals(0, ByteBuffer.wrap(qp.getData()).getInt()); | ||
| assertEquals(1, f.self.getAcceptedEpoch()); | ||
| assertEquals(0, f.self.getCurrentEpoch()); | ||
|
|
||
|
|
@@ -765,36 +765,22 @@ public void converseWithFollower(InputArchive ia, OutputArchive oa, Follower f) | |
| qp.setZxid(0); | ||
| oa.writeRecord(qp, null); | ||
|
|
||
| // Get the ack of the new leader | ||
| readPacketSkippingPing(ia, qp); | ||
| assertEquals(Leader.ACK, qp.getType()); | ||
| assertEquals(ZxidUtils.makeZxid(1, 0), qp.getZxid()); | ||
| assertEquals(1, f.self.getAcceptedEpoch()); | ||
| assertEquals(1, f.self.getCurrentEpoch()); | ||
|
There was a problem hiding this comment.
I did not see ZOOKEEPER-3023 after #2111. But if you are verifying this using TLA, this is doomed to failure. I am +1 to revert this to pre ZOOKEEPER-2678. |
||
| assertEquals(createSessionZxid, f.fzk.getLastProcessedZxid()); | ||
|
|
||
| // Read the uptodate ack | ||
| readPacketSkippingPing(ia, qp); | ||
| assertEquals(Leader.ACK, qp.getType()); | ||
| assertEquals(ZxidUtils.makeZxid(1, 0), qp.getZxid()); | ||
|
|
||
| // Make sure the data was recorded in the filesystem ok | ||
| ZKDatabase zkDb2 = new ZKDatabase(new FileTxnSnapLog(logDir, snapDir)); | ||
| zkDb2.loadDataBase(); | ||
| LOG.info("zkdb2 sessions:{}", zkDb2.getSessions()); | ||
| LOG.info("zkdb2 with timeouts:{}", zkDb2.getSessionWithTimeOuts()); | ||
| assertNotNull(zkDb2.getSessionWithTimeOuts().get(4L)); | ||
|
|
@@ -820,6 +806,140 @@ private void proposeNewSession(QuorumPacket qp, long zxid, long sessionId) throw | |
| }, testData); | ||
| } | ||
|
|
||
| @Test | ||
| public void testNormalFollowerRun_ProcessCommitInSyncAfterAckNewLeader(@TempDir File testData) throws Exception { | ||
| testFollowerConversation(new FollowerConversation() { | ||
| @Override | ||
| public void converseWithFollower(InputArchive ia, OutputArchive oa, Follower f) throws Exception { | ||
| File tmpDir = File.createTempFile("test", "dir", testData); | ||
| tmpDir.delete(); | ||
| tmpDir.mkdir(); | ||
| File logDir = f.fzk.getTxnLogFactory().getDataLogDir().getParentFile(); | ||
| File snapDir = f.fzk.getTxnLogFactory().getSnapDir().getParentFile(); | ||
| //Spy on ZK so we can check if a snapshot happened or not. | ||
| f.zk = spy(f.zk); | ||
| try { | ||
| assertEquals(0, f.self.getAcceptedEpoch()); | ||
| assertEquals(0, f.self.getCurrentEpoch()); | ||
|
|
||
| // Setup a database with a single /foo node | ||
| ZKDatabase zkDb = new ZKDatabase(new FileTxnSnapLog(tmpDir, tmpDir)); | ||
| final long firstZxid = ZxidUtils.makeZxid(1, 1); | ||
| zkDb.processTxn(new TxnHeader(13, 1313, firstZxid, 33, ZooDefs.OpCode.create), new CreateTxn("/foo", "data1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, false, 1), null); | ||
| Stat stat = new Stat(); | ||
| assertEquals("data1", new String(zkDb.getData("/foo", stat, null))); | ||
|
|
||
| QuorumPacket qp = new QuorumPacket(); | ||
| readPacketSkippingPing(ia, qp); | ||
| assertEquals(Leader.FOLLOWERINFO, qp.getType()); | ||
| assertEquals(qp.getZxid(), 0); | ||
| LearnerInfo learnInfo = new LearnerInfo(); | ||
| ByteBufferInputStream.byteBuffer2Record(ByteBuffer.wrap(qp.getData()), learnInfo); | ||
| assertEquals(learnInfo.getProtocolVersion(), 0x10000); | ||
| assertEquals(learnInfo.getServerid(), 0); | ||
|
|
||
| // We are simulating an established leader, so the epoch is 1 | ||
| qp.setType(Leader.LEADERINFO); | ||
| qp.setZxid(ZxidUtils.makeZxid(1, 0)); | ||
| byte[] protoBytes = new byte[4]; | ||
| ByteBuffer.wrap(protoBytes).putInt(0x10000); | ||
| qp.setData(protoBytes); | ||
| oa.writeRecord(qp, null); | ||
|
|
||
| readPacketSkippingPing(ia, qp); | ||
| assertEquals(Leader.ACKEPOCH, qp.getType()); | ||
| assertEquals(0, qp.getZxid()); | ||
| assertEquals(ZxidUtils.makeZxid(0, 0), ByteBuffer.wrap(qp.getData()).getInt()); | ||
| assertEquals(1, f.self.getAcceptedEpoch()); | ||
| assertEquals(0, f.self.getCurrentEpoch()); | ||
|
|
||
| // Send the snapshot we created earlier | ||
| qp.setType(Leader.SNAP); | ||
| qp.setData(new byte[0]); | ||
| qp.setZxid(zkDb.getDataTreeLastProcessedZxid()); | ||
| oa.writeRecord(qp, null); | ||
| zkDb.serializeSnapshot(oa); | ||
| oa.writeString("BenWasHere", null); | ||
| Thread.sleep(10); //Give it some time to process the snap | ||
| //No Snapshot taken yet, the SNAP was applied in memory | ||
| verify(f.zk, never()).takeSnapshot(); | ||
|
|
||
| // Leader sends an outstanding proposal | ||
| long proposalZxid = ZxidUtils.makeZxid(1, 1001); | ||
| proposeSetData(qp, proposalZxid, "data2", 2); | ||
| oa.writeRecord(qp, null); | ||
|
There was a problem hiding this comment. I think it is what ZOOKEEPER-4394 tried to report. New proposals are issued before |
||
|
|
||
| qp.setType(Leader.NEWLEADER); | ||
| qp.setZxid(ZxidUtils.makeZxid(1, 0)); | ||
| oa.writeRecord(qp, null); | ||
|
|
||
|
|
||
| // Get the ack of the new leader | ||
| readPacketSkippingPing(ia, qp); | ||
| assertEquals(Leader.ACK, qp.getType()); | ||
| assertEquals(ZxidUtils.makeZxid(1, 0), qp.getZxid()); | ||
| assertEquals(1, f.self.getAcceptedEpoch()); | ||
| assertEquals(1, f.self.getCurrentEpoch()); | ||
| //Make sure that we did take the snapshot now | ||
| verify(f.zk).takeSnapshot(true); | ||
| assertEquals(firstZxid, f.fzk.getLastProcessedZxid()); | ||
|
|
||
| // The outstanding proposal has not been persisted yet | ||
| ZKDatabase zkDb2 = new ZKDatabase(new FileTxnSnapLog(logDir, snapDir)); | ||
| long lastZxid = zkDb2.loadDataBase(); | ||
| assertEquals("data1", new String(zkDb2.getData("/foo", stat, null))); | ||
|
There was a problem hiding this comment. This seems an extra enforce to the paper. But given that we are driving the test step by step and we are testing implementation, I am +1 on this. By the paper, we should not have this problem and assertions should still hold anyway as there are no new proposals from new leader before |
||
| assertEquals(firstZxid, lastZxid); | ||
|
|
||
| TrackerWatcher watcher = new TrackerWatcher(); | ||
|
|
||
| // The change should not have happened yet | ||
| assertEquals("data1", new String(f.fzk.getZKDatabase().getData("/foo", stat, watcher))); | ||
|
|
||
| // Leader commits proposalZxid right after it sends NEWLEADER to follower | ||
| qp.setType(Leader.COMMIT); | ||
| qp.setZxid(proposalZxid); | ||
| oa.writeRecord(qp, null); | ||
|
|
||
| qp.setType(Leader.UPTODATE); | ||
| qp.setZxid(0); | ||
| oa.writeRecord(qp, null); | ||
|
|
||
| // Read the uptodate ack | ||
| readPacketSkippingPing(ia, qp); | ||
| assertEquals(Leader.ACK, qp.getType()); | ||
| assertEquals(ZxidUtils.makeZxid(1, 0), qp.getZxid()); | ||
|
|
||
| readPacketSkippingPing(ia, qp); | ||
| assertEquals(Leader.ACK, qp.getType()); | ||
| assertEquals(proposalZxid, qp.getZxid()); | ||
|
|
||
| // The change should happen now | ||
| watcher.waitForChange(); | ||
| assertEquals("data2", new String(f.fzk.getZKDatabase().getData("/foo", stat, null))); | ||
|
|
||
| // check and make sure the change is persisted | ||
| zkDb2 = new ZKDatabase(new FileTxnSnapLog(logDir, snapDir)); | ||
| lastZxid = zkDb2.loadDataBase(); | ||
| assertEquals("data2", new String(zkDb2.getData("/foo", stat, null))); | ||
| assertEquals(proposalZxid, lastZxid); | ||
| } finally { | ||
| TestUtils.deleteFileRecursively(tmpDir); | ||
| } | ||
| } | ||
|
|
||
| private void proposeSetData(QuorumPacket qp, long zxid, String data, int version) throws IOException { | ||
| qp.setType(Leader.PROPOSAL); | ||
| qp.setZxid(zxid); | ||
| TxnHeader hdr = new TxnHeader(4, 1414, qp.getZxid(), 55, ZooDefs.OpCode.setData); | ||
| SetDataTxn sdt = new SetDataTxn("/foo", data.getBytes(), version); | ||
| ByteArrayOutputStream baos = new ByteArrayOutputStream(); | ||
| OutputArchive boa = BinaryOutputArchive.getArchive(baos); | ||
| boa.writeRecord(hdr, null); | ||
| boa.writeRecord(sdt, null); | ||
| qp.setData(baos.toByteArray()); | ||
| } | ||
| }, testData); | ||
| } | ||
|
|
||
| @Test | ||
| public void testNormalRun(@TempDir File testData) throws Exception { | ||
| testLeaderConversation(new LeaderConversation() { | ||
|
|
||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The comment is somewhat misunderstanding. The key is to log these committed ones, they are considered committed before election by the paper. All the reason we touch
packetsNotCommittedhere is to make sure it is notlogRequestagain inbroadcastphase. I think it might be better to renamepacketsNotCommittedtopacketsNotLoggedas @jeffrey-xiao did in #1930. "log" is a disk operation, "commit" is an agreement. What we want here should be "log committed txns agreed in election".Coming into the implementation, new proposals could still be committed before
NEWLEADERsinceLearnerHandlerdoes not issueNEWLEADERright after these committed txns. But it does not harm us here as we are potentially to persist more but not less and new leader expect noackfor committed ones.