Skip to content

ZOOKEEPER-4394: Learner.syncWithLeader got NullPointerException #1930

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
jeffrey-xiao wants to merge 2 commits into from
Closed

ZOOKEEPER-4394: Learner.syncWithLeader got NullPointerException #1930

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
44bb60f
Add failing test demonstrating ZOOKEEPER-4394.
jeffrey-xiao Sep 30, 2022
22a0e46
ZOOKEEPER-4394: Fix NPE when receiving diff with outstanding proposal
jeffrey-xiao Sep 30, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 27 additions & 22 deletions zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Learner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -555,6 +555,7 @@ protected void syncWithLeader(long newLeaderZxid) throws Exception {
boolean syncSnapshot = false;
readPacket(qp);
Deque<Long> packetsCommitted = new ArrayDeque<>();
Deque<PacketInFlight> packetsNotLogged = new ArrayDeque<>();
Deque<PacketInFlight> packetsNotCommitted = new ArrayDeque<>();
synchronized (zk) {
if (qp.getType() == Leader.DIFF) {
Expand Down Expand Up @@ -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 {
Comment on lines +653 to +658

@anmolnar anmolnar Sep 18, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure if this branch is correct, because in the old logic if writeToTxnLog == true, then the packet is added to packetsCommitted no matter if zxid matches with the next proposal or not.

In your new logic, if I understand it right, if zxid differs you'll quit fast and log the above warning message only.

Is that correct?

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());

@anmolnar anmolnar Sep 18, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Here. This logic is not handled in the new code for some reason.

}
break;
case Leader.INFORM:
Expand Down Expand Up @@ -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());
}

Expand Down Expand Up @@ -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);
Expand All @@ -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) {
Expand All @@ -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
Expand Down
83 changes: 83 additions & 0 deletions zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LearnerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<ACL>(), 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<ACL>(), 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);
}
}
}