LearnerHandler extends ZooKeeperThread { public boolean syncFollower(long peerLastZxid, ZKDatabase db, Leader leader) { boolean isPeerNewEpochZxid = (peerLastZxid & 0xffffffffL) == 0; long currentZxid = peerLastZxid; boolean needSnap = true; boolean txnLogSyncEnabled = db.isTxnLogSyncEnabled(); ReentrantReadWriteLock lock = db.getLogLock(); ReadLock rl = lock.readLock(); try { rl.lock(); long maxCommittedLog = db.getmaxCommittedLog(); long minCommittedLog = db.getminCommittedLog(); long lastProcessedZxid = db.getDataTreeLastProcessedZxid(); LOG.info("Synchronizing with Follower sid: {} maxCommittedLog=0x{}" + " minCommittedLog=0x{} lastProcessedZxid=0x{}" + " peerLastZxid=0x{}", getSid(), Long.toHexString(maxCommittedLog), Long.toHexString(minCommittedLog), Long.toHexString(lastProcessedZxid), Long.toHexString(peerLastZxid)); if (db.getCommittedLog().isEmpty()) { minCommittedLog = lastProcessedZxid; maxCommittedLog = lastProcessedZxid; } if (forceSnapSync) { LOG.warn("Forcing snapshot sync - should not see this in production"); } else if (lastProcessedZxid == peerLastZxid) { LOG.info("Sending DIFF zxid=0x" + Long.toHexString(peerLastZxid) + " for peer sid: " + getSid()); queueOpPacket(Leader.DIFF, peerLastZxid); needOpPacket = false; needSnap = false; } else if (peerLastZxid > maxCommittedLog && !isPeerNewEpochZxid) { LOG.debug("Sending TRUNC to follower zxidToSend=0x" + Long.toHexString(maxCommittedLog) + " for peer sid:" + getSid()); queueOpPacket(Leader.TRUNC, maxCommittedLog); currentZxid = maxCommittedLog; needOpPacket = false; needSnap = false; } else if ((maxCommittedLog >= peerLastZxid) && (minCommittedLog <= peerLastZxid)) { LOG.info("Using committedLog for peer sid: " + getSid()); Iterator<Proposal> itr = db.getCommittedLog().iterator(); currentZxid = queueCommittedProposals(itr, peerLastZxid, null, maxCommittedLog); needSnap = false; } else if (peerLastZxid < minCommittedLog && txnLogSyncEnabled) { long sizeLimit = db.calculateTxnLogSizeLimit(); Iterator<Proposal> txnLogItr = db.getProposalsFromTxnLog( peerLastZxid, sizeLimit); if (txnLogItr.hasNext()) { LOG.info("Use txnlog and committedLog for peer sid: " + getSid()); currentZxid = queueCommittedProposals(txnLogItr, peerLastZxid, minCommittedLog, maxCommittedLog); LOG.debug("Queueing committedLog 0x" + Long.toHexString(currentZxid)); Iterator<Proposal> committedLogItr = db.getCommittedLog().iterator(); currentZxid = queueCommittedProposals(committedLogItr, currentZxid, null, maxCommittedLog); needSnap = false; } if (txnLogItr instanceof TxnLogProposalIterator) { TxnLogProposalIterator txnProposalItr = (TxnLogProposalIterator) txnLogItr; txnProposalItr.close(); } } else { LOG.warn("Unhandled scenario for peer sid: " + getSid()); } LOG.debug("Start forwarding 0x" + Long.toHexString(currentZxid) + " for peer sid: " + getSid()); leaderLastZxid = leader.startForwarding(this, currentZxid); } finally { rl.unlock(); } if (needOpPacket && !needSnap) { LOG.error("Unhandled scenario for peer sid: " + getSid() + " fall back to use snapshot"); needSnap = true; } return needSnap; } LearnerHandler(Socket sock, BufferedInputStream bufferedInput,Leader leader); Socket getSocket(); @Override String toString(); LearnerType getLearnerType(); static String packetToString(QuorumPacket p); @Override void run(); boolean syncFollower(long peerLastZxid, ZKDatabase db, Leader leader); void shutdown(); long tickOfNextAckDeadline(); void ping(); boolean synced(); Queue<QuorumPacket> getQueuedPackets(); void setFirstPacket(boolean value); static final String FORCE_SNAP_SYNC; }
@Test public void testTxnLogProposalIteratorClosure() throws Exception { long peerZxid; db = new MockZKDatabase(null) { @Override public Iterator<Proposal> getProposalsFromTxnLog(long peerZxid, long limit) { return TxnLogProposalIterator.EMPTY_ITERATOR; } }; db.lastProcessedZxid = 7; db.txnLog.add(createProposal(2)); db.txnLog.add(createProposal(3)); peerZxid = 4; assertTrue("Couldn't identify snapshot transfer!", learnerHandler.syncFollower(peerZxid, db, leader)); reset(); }