You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by an...@apache.org on 2018/10/05 13:36:53 UTC
[25/51] [partial] zookeeper git commit: ZOOKEEPER-3032: MAVEN
MIGRATION - branch-3.5 - move java server, client
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/3ce1b18b/src/java/main/org/apache/zookeeper/server/quorum/LeaderSessionTracker.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/LeaderSessionTracker.java b/src/java/main/org/apache/zookeeper/server/quorum/LeaderSessionTracker.java
deleted file mode 100644
index 38bbfe8..0000000
--- a/src/java/main/org/apache/zookeeper/server/quorum/LeaderSessionTracker.java
+++ /dev/null
@@ -1,221 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.zookeeper.server.quorum;
-
-import java.io.PrintWriter;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeMap;
-import java.util.concurrent.ConcurrentMap;
-
-import org.apache.zookeeper.KeeperException.SessionExpiredException;
-import org.apache.zookeeper.KeeperException.SessionMovedException;
-import org.apache.zookeeper.KeeperException.UnknownSessionException;
-import org.apache.zookeeper.server.SessionTrackerImpl;
-import org.apache.zookeeper.server.ZooKeeperServerListener;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * The leader session tracker tracks local and global sessions on the leader.
- */
-public class LeaderSessionTracker extends UpgradeableSessionTracker {
- private static final Logger LOG = LoggerFactory.getLogger(LeaderSessionTracker.class);
-
- private final boolean localSessionsEnabled;
- private final SessionTrackerImpl globalSessionTracker;
-
- /**
- * Server id of the leader
- */
- private final long serverId;
-
- public LeaderSessionTracker(SessionExpirer expirer,
- ConcurrentMap<Long, Integer> sessionsWithTimeouts,
- int tickTime, long id, boolean localSessionsEnabled,
- ZooKeeperServerListener listener) {
-
- this.globalSessionTracker = new SessionTrackerImpl(
- expirer, sessionsWithTimeouts, tickTime, id, listener);
-
- this.localSessionsEnabled = localSessionsEnabled;
- if (this.localSessionsEnabled) {
- createLocalSessionTracker(expirer, tickTime, id, listener);
- }
- serverId = id;
- }
-
- public void removeSession(long sessionId) {
- if (localSessionTracker != null) {
- localSessionTracker.removeSession(sessionId);
- }
- globalSessionTracker.removeSession(sessionId);
- }
-
- public void start() {
- globalSessionTracker.start();
- if (localSessionTracker != null) {
- localSessionTracker.start();
- }
- }
-
- public void shutdown() {
- if (localSessionTracker != null) {
- localSessionTracker.shutdown();
- }
- globalSessionTracker.shutdown();
- }
-
- public boolean isGlobalSession(long sessionId) {
- return globalSessionTracker.isTrackingSession(sessionId);
- }
-
- public boolean addGlobalSession(long sessionId, int sessionTimeout) {
- boolean added =
- globalSessionTracker.addSession(sessionId, sessionTimeout);
- if (localSessionsEnabled && added) {
- // Only do extra logging so we know what kind of session this is
- // if we're supporting both kinds of sessions
- LOG.info("Adding global session 0x" + Long.toHexString(sessionId));
- }
- return added;
- }
-
- public boolean addSession(long sessionId, int sessionTimeout) {
- boolean added;
- if (localSessionsEnabled && !isGlobalSession(sessionId)) {
- added = localSessionTracker.addSession(sessionId, sessionTimeout);
- // Check for race condition with session upgrading
- if (isGlobalSession(sessionId)) {
- added = false;
- localSessionTracker.removeSession(sessionId);
- } else if (added) {
- LOG.info("Adding local session 0x" + Long.toHexString(sessionId));
- }
- } else {
- added = addGlobalSession(sessionId, sessionTimeout);
- }
- return added;
- }
-
- public boolean touchSession(long sessionId, int sessionTimeout) {
- if (localSessionTracker != null &&
- localSessionTracker.touchSession(sessionId, sessionTimeout)) {
- return true;
- }
- return globalSessionTracker.touchSession(sessionId, sessionTimeout);
- }
-
- public long createSession(int sessionTimeout) {
- if (localSessionsEnabled) {
- return localSessionTracker.createSession(sessionTimeout);
- }
- return globalSessionTracker.createSession(sessionTimeout);
- }
-
- // Returns the serverId from the sessionId (the high order byte)
- public static long getServerIdFromSessionId(long sessionId) {
- return sessionId >> 56;
- }
-
- public void checkSession(long sessionId, Object owner)
- throws SessionExpiredException, SessionMovedException,
- UnknownSessionException {
- if (localSessionTracker != null) {
- try {
- localSessionTracker.checkSession(sessionId, owner);
- // A session can both be a local and global session during
- // upgrade
- if (!isGlobalSession(sessionId)) {
- return;
- }
- } catch(UnknownSessionException e) {
- // Ignore. We'll check instead whether it's a global session
- }
- }
- try {
- globalSessionTracker.checkSession(sessionId, owner);
- // if we can get here, it is a valid global session
- return;
- } catch (UnknownSessionException e) {
- // Ignore. This may be local session from other servers.
- }
-
- /*
- * if local session is not enabled or it used to be our local session
- * throw sessions expires
- */
- if (!localSessionsEnabled
- || (getServerIdFromSessionId(sessionId) == serverId)) {
- throw new SessionExpiredException();
- }
- }
-
- public void checkGlobalSession(long sessionId, Object owner)
- throws SessionExpiredException, SessionMovedException {
- try {
- globalSessionTracker.checkSession(sessionId, owner);
- } catch (UnknownSessionException e) {
- // For global session, if we don't know it, it is already expired
- throw new SessionExpiredException();
- }
- }
-
- public void setOwner(long sessionId, Object owner)
- throws SessionExpiredException {
- if (localSessionTracker != null) {
- try {
- localSessionTracker.setOwner(sessionId, owner);
- return;
- } catch(SessionExpiredException e) {
- // Ignore. We'll check instead whether it's a global session
- }
- }
- globalSessionTracker.setOwner(sessionId, owner);
- }
-
- public void dumpSessions(PrintWriter pwriter) {
- if (localSessionTracker != null) {
- pwriter.print("Local ");
- localSessionTracker.dumpSessions(pwriter);
- pwriter.print("Global ");
- }
- globalSessionTracker.dumpSessions(pwriter);
- }
-
- public void setSessionClosing(long sessionId) {
- // call is no-op if session isn't tracked so safe to call both
- if (localSessionTracker != null) {
- localSessionTracker.setSessionClosing(sessionId);
- }
- globalSessionTracker.setSessionClosing(sessionId);
- }
-
- public Map<Long, Set<Long>> getSessionExpiryMap() {
- Map<Long, Set<Long>> sessionExpiryMap;
- // combine local and global sessions, getting local first so upgrades
- // to global are caught
- if (localSessionTracker != null) {
- sessionExpiryMap = localSessionTracker.getSessionExpiryMap();
- } else {
- sessionExpiryMap = new TreeMap<Long, Set<Long>>();
- }
- sessionExpiryMap.putAll(globalSessionTracker.getSessionExpiryMap());
- return sessionExpiryMap;
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/3ce1b18b/src/java/main/org/apache/zookeeper/server/quorum/LeaderZooKeeperServer.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/LeaderZooKeeperServer.java b/src/java/main/org/apache/zookeeper/server/quorum/LeaderZooKeeperServer.java
deleted file mode 100644
index 4f8c095..0000000
--- a/src/java/main/org/apache/zookeeper/server/quorum/LeaderZooKeeperServer.java
+++ /dev/null
@@ -1,238 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server.quorum;
-
-import org.apache.zookeeper.KeeperException.SessionExpiredException;
-import org.apache.zookeeper.jmx.MBeanRegistry;
-import org.apache.zookeeper.server.ContainerManager;
-import org.apache.zookeeper.server.DataTreeBean;
-import org.apache.zookeeper.server.FinalRequestProcessor;
-import org.apache.zookeeper.server.PrepRequestProcessor;
-import org.apache.zookeeper.server.Request;
-import org.apache.zookeeper.server.RequestProcessor;
-import org.apache.zookeeper.server.ServerCnxn;
-import org.apache.zookeeper.server.ZKDatabase;
-import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
-
-import java.io.IOException;
-import java.util.concurrent.TimeUnit;
-
-/**
- *
- * Just like the standard ZooKeeperServer. We just replace the request
- * processors: PrepRequestProcessor -> ProposalRequestProcessor ->
- * CommitProcessor -> Leader.ToBeAppliedRequestProcessor ->
- * FinalRequestProcessor
- */
-public class LeaderZooKeeperServer extends QuorumZooKeeperServer {
- private ContainerManager containerManager; // guarded by sync
-
-
- CommitProcessor commitProcessor;
-
- PrepRequestProcessor prepRequestProcessor;
-
- /**
- * @param port
- * @param dataDir
- * @throws IOException
- */
- LeaderZooKeeperServer(FileTxnSnapLog logFactory, QuorumPeer self, ZKDatabase zkDb) throws IOException {
- super(logFactory, self.tickTime, self.minSessionTimeout, self.maxSessionTimeout, zkDb, self);
- }
-
- public Leader getLeader(){
- return self.leader;
- }
-
- @Override
- protected void setupRequestProcessors() {
- RequestProcessor finalProcessor = new FinalRequestProcessor(this);
- RequestProcessor toBeAppliedProcessor = new Leader.ToBeAppliedRequestProcessor(finalProcessor, getLeader());
- commitProcessor = new CommitProcessor(toBeAppliedProcessor,
- Long.toString(getServerId()), false,
- getZooKeeperServerListener());
- commitProcessor.start();
- ProposalRequestProcessor proposalProcessor = new ProposalRequestProcessor(this,
- commitProcessor);
- proposalProcessor.initialize();
- prepRequestProcessor = new PrepRequestProcessor(this, proposalProcessor);
- prepRequestProcessor.start();
- firstProcessor = new LeaderRequestProcessor(this, prepRequestProcessor);
-
- setupContainerManager();
- }
-
- private synchronized void setupContainerManager() {
- containerManager = new ContainerManager(getZKDatabase(), prepRequestProcessor,
- Integer.getInteger("znode.container.checkIntervalMs", (int) TimeUnit.MINUTES.toMillis(1)),
- Integer.getInteger("znode.container.maxPerMinute", 10000)
- );
- }
-
- @Override
- public synchronized void startup() {
- super.startup();
- if (containerManager != null) {
- containerManager.start();
- }
- }
-
- @Override
- public synchronized void shutdown() {
- if (containerManager != null) {
- containerManager.stop();
- }
- super.shutdown();
- }
-
- @Override
- public int getGlobalOutstandingLimit() {
- int divisor = self.getQuorumSize() > 2 ? self.getQuorumSize() - 1 : 1;
- return super.getGlobalOutstandingLimit() / divisor;
- }
-
- @Override
- public void createSessionTracker() {
- sessionTracker = new LeaderSessionTracker(
- this, getZKDatabase().getSessionWithTimeOuts(),
- tickTime, self.getId(), self.areLocalSessionsEnabled(),
- getZooKeeperServerListener());
- }
-
- public boolean touch(long sess, int to) {
- return sessionTracker.touchSession(sess, to);
- }
-
- public boolean checkIfValidGlobalSession(long sess, int to) {
- if (self.areLocalSessionsEnabled() &&
- !upgradeableSessionTracker.isGlobalSession(sess)) {
- return false;
- }
- return sessionTracker.touchSession(sess, to);
- }
-
- /**
- * Requests coming from the learner should go directly to
- * PrepRequestProcessor
- *
- * @param request
- */
- public void submitLearnerRequest(Request request) {
- /*
- * Requests coming from the learner should have gone through
- * submitRequest() on each server which already perform some request
- * validation, so we don't need to do it again.
- *
- * Additionally, LearnerHandler should start submitting requests into
- * the leader's pipeline only when the leader's server is started, so we
- * can submit the request directly into PrepRequestProcessor.
- *
- * This is done so that requests from learners won't go through
- * LeaderRequestProcessor which perform local session upgrade.
- */
- prepRequestProcessor.processRequest(request);
- }
-
- @Override
- protected void registerJMX() {
- // register with JMX
- try {
- jmxDataTreeBean = new DataTreeBean(getZKDatabase().getDataTree());
- MBeanRegistry.getInstance().register(jmxDataTreeBean, jmxServerBean);
- } catch (Exception e) {
- LOG.warn("Failed to register with JMX", e);
- jmxDataTreeBean = null;
- }
- }
-
- public void registerJMX(LeaderBean leaderBean,
- LocalPeerBean localPeerBean)
- {
- // register with JMX
- if (self.jmxLeaderElectionBean != null) {
- try {
- MBeanRegistry.getInstance().unregister(self.jmxLeaderElectionBean);
- } catch (Exception e) {
- LOG.warn("Failed to register with JMX", e);
- }
- self.jmxLeaderElectionBean = null;
- }
-
- try {
- jmxServerBean = leaderBean;
- MBeanRegistry.getInstance().register(leaderBean, localPeerBean);
- } catch (Exception e) {
- LOG.warn("Failed to register with JMX", e);
- jmxServerBean = null;
- }
- }
-
- @Override
- protected void unregisterJMX() {
- // unregister from JMX
- try {
- if (jmxDataTreeBean != null) {
- MBeanRegistry.getInstance().unregister(jmxDataTreeBean);
- }
- } catch (Exception e) {
- LOG.warn("Failed to unregister with JMX", e);
- }
- jmxDataTreeBean = null;
- }
-
- protected void unregisterJMX(Leader leader) {
- // unregister from JMX
- try {
- if (jmxServerBean != null) {
- MBeanRegistry.getInstance().unregister(jmxServerBean);
- }
- } catch (Exception e) {
- LOG.warn("Failed to unregister with JMX", e);
- }
- jmxServerBean = null;
- }
-
- @Override
- public String getState() {
- return "leader";
- }
-
- /**
- * Returns the id of the associated QuorumPeer, which will do for a unique
- * id of this server.
- */
- @Override
- public long getServerId() {
- return self.getId();
- }
-
- @Override
- protected void revalidateSession(ServerCnxn cnxn, long sessionId,
- int sessionTimeout) throws IOException {
- super.revalidateSession(cnxn, sessionId, sessionTimeout);
- try {
- // setowner as the leader itself, unless updated
- // via the follower handlers
- setOwner(sessionId, ServerCnxn.me);
- } catch (SessionExpiredException e) {
- // this is ok, it just means that the session revalidation failed.
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/3ce1b18b/src/java/main/org/apache/zookeeper/server/quorum/Learner.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/Learner.java b/src/java/main/org/apache/zookeeper/server/quorum/Learner.java
deleted file mode 100644
index 667f73c..0000000
--- a/src/java/main/org/apache/zookeeper/server/quorum/Learner.java
+++ /dev/null
@@ -1,677 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.zookeeper.server.quorum;
-
-import java.io.BufferedInputStream;
-import java.io.BufferedOutputStream;
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.net.ConnectException;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.nio.ByteBuffer;
-import java.util.LinkedList;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.apache.jute.BinaryInputArchive;
-import org.apache.jute.BinaryOutputArchive;
-import org.apache.jute.InputArchive;
-import org.apache.jute.OutputArchive;
-import org.apache.jute.Record;
-import org.apache.zookeeper.common.QuorumX509Util;
-import org.apache.zookeeper.common.X509Exception;
-import org.apache.zookeeper.common.X509Util;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.zookeeper.ZooDefs.OpCode;
-import org.apache.zookeeper.server.Request;
-import org.apache.zookeeper.server.ServerCnxn;
-import org.apache.zookeeper.server.ZooTrace;
-import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
-import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
-import org.apache.zookeeper.server.util.SerializeUtils;
-import org.apache.zookeeper.server.util.ZxidUtils;
-import org.apache.zookeeper.txn.SetDataTxn;
-import org.apache.zookeeper.txn.TxnHeader;
-
-import javax.net.ssl.SSLSocket;
-
-/**
- * This class is the superclass of two of the three main actors in a ZK
- * ensemble: Followers and Observers. Both Followers and Observers share
- * a good deal of code which is moved into Peer to avoid duplication.
- */
-public class Learner {
- static class PacketInFlight {
- TxnHeader hdr;
- Record rec;
- }
- QuorumPeer self;
- LearnerZooKeeperServer zk;
-
- protected BufferedOutputStream bufferedOutput;
-
- protected Socket sock;
-
- protected X509Util x509Util;
-
- /**
- * Socket getter
- * @return
- */
- public Socket getSocket() {
- return sock;
- }
-
- protected InputArchive leaderIs;
- protected OutputArchive leaderOs;
- /** the protocol version of the leader */
- protected int leaderProtocolVersion = 0x01;
-
- protected static final Logger LOG = LoggerFactory.getLogger(Learner.class);
-
- static final private boolean nodelay = System.getProperty("follower.nodelay", "true").equals("true");
- static {
- LOG.info("TCP NoDelay set to: " + nodelay);
- }
-
- final ConcurrentHashMap<Long, ServerCnxn> pendingRevalidations =
- new ConcurrentHashMap<Long, ServerCnxn>();
-
- public int getPendingRevalidationsCount() {
- return pendingRevalidations.size();
- }
-
- /**
- * validate a session for a client
- *
- * @param clientId
- * the client to be revalidated
- * @param timeout
- * the timeout for which the session is valid
- * @return
- * @throws IOException
- */
- void validateSession(ServerCnxn cnxn, long clientId, int timeout)
- throws IOException {
- LOG.info("Revalidating client: 0x" + Long.toHexString(clientId));
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- DataOutputStream dos = new DataOutputStream(baos);
- dos.writeLong(clientId);
- dos.writeInt(timeout);
- dos.close();
- QuorumPacket qp = new QuorumPacket(Leader.REVALIDATE, -1, baos
- .toByteArray(), null);
- pendingRevalidations.put(clientId, cnxn);
- if (LOG.isTraceEnabled()) {
- ZooTrace.logTraceMessage(LOG,
- ZooTrace.SESSION_TRACE_MASK,
- "To validate session 0x"
- + Long.toHexString(clientId));
- }
- writePacket(qp, true);
- }
-
- /**
- * write a packet to the leader
- *
- * @param pp
- * the proposal packet to be sent to the leader
- * @throws IOException
- */
- void writePacket(QuorumPacket pp, boolean flush) throws IOException {
- synchronized (leaderOs) {
- if (pp != null) {
- leaderOs.writeRecord(pp, "packet");
- }
- if (flush) {
- bufferedOutput.flush();
- }
- }
- }
-
- /**
- * read a packet from the leader
- *
- * @param pp
- * the packet to be instantiated
- * @throws IOException
- */
- void readPacket(QuorumPacket pp) throws IOException {
- synchronized (leaderIs) {
- leaderIs.readRecord(pp, "packet");
- }
- long traceMask = ZooTrace.SERVER_PACKET_TRACE_MASK;
- if (pp.getType() == Leader.PING) {
- traceMask = ZooTrace.SERVER_PING_TRACE_MASK;
- }
- if (LOG.isTraceEnabled()) {
- ZooTrace.logQuorumPacket(LOG, traceMask, 'i', pp);
- }
- }
-
- /**
- * send a request packet to the leader
- *
- * @param request
- * the request from the client
- * @throws IOException
- */
- void request(Request request) throws IOException {
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- DataOutputStream oa = new DataOutputStream(baos);
- oa.writeLong(request.sessionId);
- oa.writeInt(request.cxid);
- oa.writeInt(request.type);
- if (request.request != null) {
- request.request.rewind();
- int len = request.request.remaining();
- byte b[] = new byte[len];
- request.request.get(b);
- request.request.rewind();
- oa.write(b);
- }
- oa.close();
- QuorumPacket qp = new QuorumPacket(Leader.REQUEST, -1, baos
- .toByteArray(), request.authInfo);
- writePacket(qp, true);
- }
-
- /**
- * Returns the address of the node we think is the leader.
- */
- protected QuorumServer findLeader() {
- QuorumServer leaderServer = null;
- // Find the leader by id
- Vote current = self.getCurrentVote();
- for (QuorumServer s : self.getView().values()) {
- if (s.id == current.getId()) {
- // Ensure we have the leader's correct IP address before
- // attempting to connect.
- s.recreateSocketAddresses();
- leaderServer = s;
- break;
- }
- }
- if (leaderServer == null) {
- LOG.warn("Couldn't find the leader with id = "
- + current.getId());
- }
- return leaderServer;
- }
-
- /**
- * Overridable helper method to return the System.nanoTime().
- * This method behaves identical to System.nanoTime().
- */
- protected long nanoTime() {
- return System.nanoTime();
- }
-
- /**
- * Overridable helper method to simply call sock.connect(). This can be
- * overriden in tests to fake connection success/failure for connectToLeader.
- */
- protected void sockConnect(Socket sock, InetSocketAddress addr, int timeout)
- throws IOException {
- sock.connect(addr, timeout);
- }
-
- /**
- * Establish a connection with the Leader found by findLeader. Retries
- * until either initLimit time has elapsed or 5 tries have happened.
- * @param addr - the address of the Leader to connect to.
- * @throws IOException - if the socket connection fails on the 5th attempt
- * <li>if there is an authentication failure while connecting to leader</li>
- * @throws ConnectException
- * @throws InterruptedException
- */
- protected void connectToLeader(InetSocketAddress addr, String hostname)
- throws IOException, InterruptedException, X509Exception {
- this.sock = createSocket();
-
- int initLimitTime = self.tickTime * self.initLimit;
- int remainingInitLimitTime = initLimitTime;
- long startNanoTime = nanoTime();
-
- for (int tries = 0; tries < 5; tries++) {
- try {
- // recalculate the init limit time because retries sleep for 1000 milliseconds
- remainingInitLimitTime = initLimitTime - (int)((nanoTime() - startNanoTime) / 1000000);
- if (remainingInitLimitTime <= 0) {
- LOG.error("initLimit exceeded on retries.");
- throw new IOException("initLimit exceeded on retries.");
- }
-
- sockConnect(sock, addr, Math.min(self.tickTime * self.syncLimit, remainingInitLimitTime));
- if (self.isSslQuorum()) {
- ((SSLSocket) sock).startHandshake();
- }
- sock.setTcpNoDelay(nodelay);
- break;
- } catch (IOException e) {
- remainingInitLimitTime = initLimitTime - (int)((nanoTime() - startNanoTime) / 1000000);
-
- if (remainingInitLimitTime <= 1000) {
- LOG.error("Unexpected exception, initLimit exceeded. tries=" + tries +
- ", remaining init limit=" + remainingInitLimitTime +
- ", connecting to " + addr,e);
- throw e;
- } else if (tries >= 4) {
- LOG.error("Unexpected exception, retries exceeded. tries=" + tries +
- ", remaining init limit=" + remainingInitLimitTime +
- ", connecting to " + addr,e);
- throw e;
- } else {
- LOG.warn("Unexpected exception, tries=" + tries +
- ", remaining init limit=" + remainingInitLimitTime +
- ", connecting to " + addr,e);
- this.sock = createSocket();
- }
- }
- Thread.sleep(1000);
- }
-
- self.authLearner.authenticate(sock, hostname);
-
- leaderIs = BinaryInputArchive.getArchive(new BufferedInputStream(
- sock.getInputStream()));
- bufferedOutput = new BufferedOutputStream(sock.getOutputStream());
- leaderOs = BinaryOutputArchive.getArchive(bufferedOutput);
- }
-
- private Socket createSocket() throws X509Exception, IOException {
- Socket sock;
- if (self.isSslQuorum()) {
- if (x509Util == null) {
- x509Util = new QuorumX509Util();
- }
- sock = x509Util.createSSLSocket();
- } else {
- sock = new Socket();
- }
- sock.setSoTimeout(self.tickTime * self.initLimit);
- return sock;
- }
-
- /**
- * Once connected to the leader, perform the handshake protocol to
- * establish a following / observing connection.
- * @param pktType
- * @return the zxid the Leader sends for synchronization purposes.
- * @throws IOException
- */
- protected long registerWithLeader(int pktType) throws IOException{
- /*
- * Send follower info, including last zxid and sid
- */
- long lastLoggedZxid = self.getLastLoggedZxid();
- QuorumPacket qp = new QuorumPacket();
- qp.setType(pktType);
- qp.setZxid(ZxidUtils.makeZxid(self.getAcceptedEpoch(), 0));
-
- /*
- * Add sid to payload
- */
- LearnerInfo li = new LearnerInfo(self.getId(), 0x10000, self.getQuorumVerifier().getVersion());
- ByteArrayOutputStream bsid = new ByteArrayOutputStream();
- BinaryOutputArchive boa = BinaryOutputArchive.getArchive(bsid);
- boa.writeRecord(li, "LearnerInfo");
- qp.setData(bsid.toByteArray());
-
- writePacket(qp, true);
- readPacket(qp);
- final long newEpoch = ZxidUtils.getEpochFromZxid(qp.getZxid());
- if (qp.getType() == Leader.LEADERINFO) {
- // we are connected to a 1.0 server so accept the new epoch and read the next packet
- leaderProtocolVersion = ByteBuffer.wrap(qp.getData()).getInt();
- byte epochBytes[] = new byte[4];
- final ByteBuffer wrappedEpochBytes = ByteBuffer.wrap(epochBytes);
- if (newEpoch > self.getAcceptedEpoch()) {
- wrappedEpochBytes.putInt((int)self.getCurrentEpoch());
- self.setAcceptedEpoch(newEpoch);
- } else if (newEpoch == self.getAcceptedEpoch()) {
- // since we have already acked an epoch equal to the leaders, we cannot ack
- // again, but we still need to send our lastZxid to the leader so that we can
- // sync with it if it does assume leadership of the epoch.
- // the -1 indicates that this reply should not count as an ack for the new epoch
- wrappedEpochBytes.putInt(-1);
- } else {
- throw new IOException("Leaders epoch, " + newEpoch + " is less than accepted epoch, " + self.getAcceptedEpoch());
- }
- QuorumPacket ackNewEpoch = new QuorumPacket(Leader.ACKEPOCH, lastLoggedZxid, epochBytes, null);
- writePacket(ackNewEpoch, true);
- return ZxidUtils.makeZxid(newEpoch, 0);
- } else {
- if (newEpoch > self.getAcceptedEpoch()) {
- self.setAcceptedEpoch(newEpoch);
- }
- if (qp.getType() != Leader.NEWLEADER) {
- LOG.error("First packet should have been NEWLEADER");
- throw new IOException("First packet should have been NEWLEADER");
- }
- return qp.getZxid();
- }
- }
-
- /**
- * Finally, synchronize our history with the Leader.
- * @param newLeaderZxid
- * @throws IOException
- * @throws InterruptedException
- */
- protected void syncWithLeader(long newLeaderZxid) throws Exception{
- QuorumPacket ack = new QuorumPacket(Leader.ACK, 0, null, null);
- QuorumPacket qp = new QuorumPacket();
- long newEpoch = ZxidUtils.getEpochFromZxid(newLeaderZxid);
-
- QuorumVerifier newLeaderQV = null;
-
- // In the DIFF case we don't need to do a snapshot because the transactions will sync on top of any existing snapshot
- // For SNAP and TRUNC the snapshot is needed to save that history
- boolean snapshotNeeded = true;
- readPacket(qp);
- LinkedList<Long> packetsCommitted = new LinkedList<Long>();
- LinkedList<PacketInFlight> packetsNotCommitted = new LinkedList<PacketInFlight>();
- synchronized (zk) {
- if (qp.getType() == Leader.DIFF) {
- LOG.info("Getting a diff from the leader 0x{}", Long.toHexString(qp.getZxid()));
- snapshotNeeded = false;
- }
- else if (qp.getType() == Leader.SNAP) {
- LOG.info("Getting a snapshot from leader 0x" + Long.toHexString(qp.getZxid()));
- // The leader is going to dump the database
- // db is clear as part of deserializeSnapshot()
- zk.getZKDatabase().deserializeSnapshot(leaderIs);
- // ZOOKEEPER-2819: overwrite config node content extracted
- // from leader snapshot with local config, to avoid potential
- // inconsistency of config node content during rolling restart.
- if (!QuorumPeerConfig.isReconfigEnabled()) {
- LOG.debug("Reset config node content from local config after deserialization of snapshot.");
- zk.getZKDatabase().initConfigInZKDatabase(self.getQuorumVerifier());
- }
- String signature = leaderIs.readString("signature");
- if (!signature.equals("BenWasHere")) {
- LOG.error("Missing signature. Got " + signature);
- throw new IOException("Missing signature");
- }
- zk.getZKDatabase().setlastProcessedZxid(qp.getZxid());
- } else if (qp.getType() == Leader.TRUNC) {
- //we need to truncate the log to the lastzxid of the leader
- LOG.warn("Truncating log to get in sync with the leader 0x"
- + Long.toHexString(qp.getZxid()));
- boolean truncated=zk.getZKDatabase().truncateLog(qp.getZxid());
- if (!truncated) {
- // not able to truncate the log
- LOG.error("Not able to truncate the log "
- + Long.toHexString(qp.getZxid()));
- System.exit(13);
- }
- zk.getZKDatabase().setlastProcessedZxid(qp.getZxid());
-
- }
- else {
- LOG.error("Got unexpected packet from leader: {}, exiting ... ",
- LearnerHandler.packetToString(qp));
- System.exit(13);
-
- }
- zk.getZKDatabase().initConfigInZKDatabase(self.getQuorumVerifier());
- zk.createSessionTracker();
-
- long lastQueued = 0;
-
- // in Zab V1.0 (ZK 3.4+) we might take a snapshot when we get the NEWLEADER message, but in pre V1.0
- // we take the snapshot on the UPDATE message, since Zab V1.0 also gets the UPDATE (after the NEWLEADER)
- // we need to make sure that we don't take the snapshot twice.
- boolean isPreZAB1_0 = true;
- //If we are not going to take the snapshot be sure the transactions are not applied in memory
- // but written out to the transaction log
- boolean writeToTxnLog = !snapshotNeeded;
- // we are now going to start getting transactions to apply followed by an UPTODATE
- outerLoop:
- while (self.isRunning()) {
- readPacket(qp);
- switch(qp.getType()) {
- case Leader.PROPOSAL:
- PacketInFlight pif = new PacketInFlight();
- pif.hdr = new TxnHeader();
- pif.rec = SerializeUtils.deserializeTxn(qp.getData(), pif.hdr);
- if (pif.hdr.getZxid() != lastQueued + 1) {
- LOG.warn("Got zxid 0x"
- + Long.toHexString(pif.hdr.getZxid())
- + " expected 0x"
- + Long.toHexString(lastQueued + 1));
- }
- lastQueued = pif.hdr.getZxid();
-
- if (pif.hdr.getType() == OpCode.reconfig){
- SetDataTxn setDataTxn = (SetDataTxn) pif.rec;
- QuorumVerifier qv = self.configFromString(new String(setDataTxn.getData()));
- self.setLastSeenQuorumVerifier(qv, true);
- }
-
- 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()));
- 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 " + qp.getZxid() + ", but next proposal is " + pif.hdr.getZxid());
- } else {
- zk.processTxn(pif.hdr, pif.rec);
- packetsNotCommitted.remove();
- }
- } else {
- packetsCommitted.add(qp.getZxid());
- }
- break;
- case Leader.INFORM:
- case Leader.INFORMANDACTIVATE:
- PacketInFlight packet = new PacketInFlight();
- packet.hdr = new TxnHeader();
-
- if (qp.getType() == Leader.INFORMANDACTIVATE) {
- ByteBuffer buffer = ByteBuffer.wrap(qp.getData());
- long suggestedLeaderId = buffer.getLong();
- byte[] remainingdata = new byte[buffer.remaining()];
- buffer.get(remainingdata);
- packet.rec = SerializeUtils.deserializeTxn(remainingdata, packet.hdr);
- QuorumVerifier qv = self.configFromString(new String(((SetDataTxn)packet.rec).getData()));
- boolean majorChange =
- self.processReconfig(qv, suggestedLeaderId, qp.getZxid(), true);
- if (majorChange) {
- throw new Exception("changes proposed in reconfig");
- }
- } else {
- packet.rec = SerializeUtils.deserializeTxn(qp.getData(), packet.hdr);
- // Log warning message if txn comes out-of-order
- if (packet.hdr.getZxid() != lastQueued + 1) {
- LOG.warn("Got zxid 0x"
- + Long.toHexString(packet.hdr.getZxid())
- + " expected 0x"
- + Long.toHexString(lastQueued + 1));
- }
- lastQueued = packet.hdr.getZxid();
- }
- if (!writeToTxnLog) {
- // Apply to db directly if we haven't taken the snapshot
- zk.processTxn(packet.hdr, packet.rec);
- } else {
- packetsNotCommitted.add(packet);
- packetsCommitted.add(qp.getZxid());
- }
-
- break;
- case Leader.UPTODATE:
- LOG.info("Learner received UPTODATE message");
- if (newLeaderQV!=null) {
- boolean majorChange =
- self.processReconfig(newLeaderQV, null, null, true);
- if (majorChange) {
- throw new Exception("changes proposed in reconfig");
- }
- }
- if (isPreZAB1_0) {
- zk.takeSnapshot();
- self.setCurrentEpoch(newEpoch);
- }
- self.setZooKeeperServer(zk);
- self.adminServer.setZooKeeperServer(zk);
- break outerLoop;
- case Leader.NEWLEADER: // Getting NEWLEADER here instead of in discovery
- // means this is Zab 1.0
- LOG.info("Learner received NEWLEADER message");
- if (qp.getData()!=null && qp.getData().length > 1) {
- try {
- QuorumVerifier qv = self.configFromString(new String(qp.getData()));
- self.setLastSeenQuorumVerifier(qv, true);
- newLeaderQV = qv;
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
-
- if (snapshotNeeded) {
- zk.takeSnapshot();
- }
-
- self.setCurrentEpoch(newEpoch);
- writeToTxnLog = true; //Anything after this needs to go to the transaction log, not applied directly in memory
- isPreZAB1_0 = false;
- writePacket(new QuorumPacket(Leader.ACK, newLeaderZxid, null, null), true);
- break;
- }
- }
- }
- ack.setZxid(ZxidUtils.makeZxid(newEpoch, 0));
- writePacket(ack, true);
- sock.setSoTimeout(self.tickTime * self.syncLimit);
- zk.startup();
- /*
- * Update the election vote here to ensure that all members of the
- * ensemble report the same vote to new servers that start up and
- * send leader election notifications to the ensemble.
- *
- * @see https://issues.apache.org/jira/browse/ZOOKEEPER-1732
- */
- self.updateElectionVote(newEpoch);
-
- // 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) {
- fzk.logRequest(p.hdr, p.rec);
- }
- for(Long zxid: packetsCommitted) {
- fzk.commit(zxid);
- }
- } else if (zk instanceof ObserverZooKeeperServer) {
- // Similar to follower, we need to log requests between the snapshot
- // and UPTODATE
- ObserverZooKeeperServer ozk = (ObserverZooKeeperServer) zk;
- for (PacketInFlight p : packetsNotCommitted) {
- Long zxid = packetsCommitted.peekFirst();
- if (p.hdr.getZxid() != zxid) {
- // log warning message if there is no matching commit
- // old leader send outstanding proposal to observer
- LOG.warn("Committing " + Long.toHexString(zxid)
- + ", but next proposal is "
- + Long.toHexString(p.hdr.getZxid()));
- continue;
- }
- packetsCommitted.remove();
- Request request = new Request(null, p.hdr.getClientId(),
- p.hdr.getCxid(), p.hdr.getType(), null, null);
- request.setTxn(p.rec);
- request.setHdr(p.hdr);
- ozk.commitRequest(request);
- }
- } else {
- // New server type need to handle in-flight packets
- throw new UnsupportedOperationException("Unknown server type");
- }
- }
-
- protected void revalidate(QuorumPacket qp) throws IOException {
- ByteArrayInputStream bis = new ByteArrayInputStream(qp
- .getData());
- DataInputStream dis = new DataInputStream(bis);
- long sessionId = dis.readLong();
- boolean valid = dis.readBoolean();
- ServerCnxn cnxn = pendingRevalidations.remove(sessionId);
- if (cnxn == null) {
- LOG.warn("Missing session 0x"
- + Long.toHexString(sessionId)
- + " for validation");
- } else {
- zk.finishSessionInit(cnxn, valid);
- }
- if (LOG.isTraceEnabled()) {
- ZooTrace.logTraceMessage(LOG,
- ZooTrace.SESSION_TRACE_MASK,
- "Session 0x" + Long.toHexString(sessionId)
- + " is valid: " + valid);
- }
- }
-
- protected void ping(QuorumPacket qp) throws IOException {
- // Send back the ping with our session data
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- DataOutputStream dos = new DataOutputStream(bos);
- Map<Long, Integer> touchTable = zk.getTouchSnapshot();
- for (Entry<Long, Integer> entry : touchTable.entrySet()) {
- dos.writeLong(entry.getKey());
- dos.writeInt(entry.getValue());
- }
- qp.setData(bos.toByteArray());
- writePacket(qp, true);
- }
-
-
- /**
- * Shutdown the Peer
- */
- public void shutdown() {
- self.setZooKeeperServer(null);
- self.closeAllConnections();
- self.adminServer.setZooKeeperServer(null);
- // shutdown previous zookeeper
- if (zk != null) {
- zk.shutdown();
- }
- }
-
- boolean isRunning() {
- return self.isRunning() && zk.isRunning();
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/3ce1b18b/src/java/main/org/apache/zookeeper/server/quorum/LearnerHandler.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/LearnerHandler.java b/src/java/main/org/apache/zookeeper/server/quorum/LearnerHandler.java
deleted file mode 100644
index f6c68b0..0000000
--- a/src/java/main/org/apache/zookeeper/server/quorum/LearnerHandler.java
+++ /dev/null
@@ -1,995 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server.quorum;
-
-import java.io.BufferedInputStream;
-import java.io.BufferedOutputStream;
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.net.Socket;
-import java.nio.ByteBuffer;
-import java.util.Iterator;
-import java.util.Queue;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
-
-import javax.security.sasl.SaslException;
-
-import org.apache.jute.BinaryInputArchive;
-import org.apache.jute.BinaryOutputArchive;
-import org.apache.jute.Record;
-import org.apache.zookeeper.KeeperException.SessionExpiredException;
-import org.apache.zookeeper.ZooDefs.OpCode;
-import org.apache.zookeeper.server.Request;
-import org.apache.zookeeper.server.TxnLogProposalIterator;
-import org.apache.zookeeper.server.ZKDatabase;
-import org.apache.zookeeper.server.ZooKeeperThread;
-import org.apache.zookeeper.server.ZooTrace;
-import org.apache.zookeeper.server.quorum.Leader.Proposal;
-import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
-import org.apache.zookeeper.server.util.SerializeUtils;
-import org.apache.zookeeper.server.util.ZxidUtils;
-import org.apache.zookeeper.txn.TxnHeader;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * There will be an instance of this class created by the Leader for each
- * learner. All communication with a learner is handled by this
- * class.
- */
-public class LearnerHandler extends ZooKeeperThread {
- private static final Logger LOG = LoggerFactory.getLogger(LearnerHandler.class);
-
- protected final Socket sock;
-
- public Socket getSocket() {
- return sock;
- }
-
- final Leader leader;
-
- /** Deadline for receiving the next ack. If we are bootstrapping then
- * it's based on the initLimit, if we are done bootstrapping it's based
- * on the syncLimit. Once the deadline is past this learner should
- * be considered no longer "sync'd" with the leader. */
- volatile long tickOfNextAckDeadline;
-
- /**
- * ZooKeeper server identifier of this learner
- */
- protected long sid = 0;
-
- long getSid(){
- return sid;
- }
-
- protected int version = 0x1;
-
- int getVersion() {
- return version;
- }
-
- /**
- * The packets to be sent to the learner
- */
- final LinkedBlockingQueue<QuorumPacket> queuedPackets =
- new LinkedBlockingQueue<QuorumPacket>();
-
- /**
- * This class controls the time that the Leader has been
- * waiting for acknowledgement of a proposal from this Learner.
- * If the time is above syncLimit, the connection will be closed.
- * It keeps track of only one proposal at a time, when the ACK for
- * that proposal arrives, it switches to the last proposal received
- * or clears the value if there is no pending proposal.
- */
- private class SyncLimitCheck {
- private boolean started = false;
- private long currentZxid = 0;
- private long currentTime = 0;
- private long nextZxid = 0;
- private long nextTime = 0;
-
- public synchronized void start() {
- started = true;
- }
-
- public synchronized void updateProposal(long zxid, long time) {
- if (!started) {
- return;
- }
- if (currentTime == 0) {
- currentTime = time;
- currentZxid = zxid;
- } else {
- nextTime = time;
- nextZxid = zxid;
- }
- }
-
- public synchronized void updateAck(long zxid) {
- if (currentZxid == zxid) {
- currentTime = nextTime;
- currentZxid = nextZxid;
- nextTime = 0;
- nextZxid = 0;
- } else if (nextZxid == zxid) {
- LOG.warn("ACK for " + zxid + " received before ACK for " + currentZxid + "!!!!");
- nextTime = 0;
- nextZxid = 0;
- }
- }
-
- public synchronized boolean check(long time) {
- if (currentTime == 0) {
- return true;
- } else {
- long msDelay = (time - currentTime) / 1000000;
- return (msDelay < (leader.self.tickTime * leader.self.syncLimit));
- }
- }
- };
-
- private SyncLimitCheck syncLimitCheck = new SyncLimitCheck();
-
- private BinaryInputArchive ia;
-
- private BinaryOutputArchive oa;
-
- private final BufferedInputStream bufferedInput;
- private BufferedOutputStream bufferedOutput;
-
- /**
- * Keep track of whether we have started send packets thread
- */
- private volatile boolean sendingThreadStarted = false;
-
- /**
- * For testing purpose, force leader to use snapshot to sync with followers
- */
- public static final String FORCE_SNAP_SYNC = "zookeeper.forceSnapshotSync";
- private boolean forceSnapSync = false;
-
- /**
- * Keep track of whether we need to queue TRUNC or DIFF into packet queue
- * that we are going to blast it to the learner
- */
- private boolean needOpPacket = true;
-
- /**
- * Last zxid sent to the learner as part of synchronization
- */
- private long leaderLastZxid;
-
- LearnerHandler(Socket sock, BufferedInputStream bufferedInput,Leader leader) throws IOException {
- super("LearnerHandler-" + sock.getRemoteSocketAddress());
- this.sock = sock;
- this.leader = leader;
- this.bufferedInput = bufferedInput;
-
- try {
- if (leader.self != null) {
- leader.self.authServer.authenticate(sock,
- new DataInputStream(bufferedInput));
- }
- } catch (IOException e) {
- LOG.error("Server failed to authenticate quorum learner, addr: {}, closing connection",
- sock.getRemoteSocketAddress(), e);
- try {
- sock.close();
- } catch (IOException ie) {
- LOG.error("Exception while closing socket", ie);
- }
- throw new SaslException("Authentication failure: " + e.getMessage());
- }
- }
-
- @Override
- public String toString() {
- StringBuilder sb = new StringBuilder();
- sb.append("LearnerHandler ").append(sock);
- sb.append(" tickOfNextAckDeadline:").append(tickOfNextAckDeadline());
- sb.append(" synced?:").append(synced());
- sb.append(" queuedPacketLength:").append(queuedPackets.size());
- return sb.toString();
- }
-
- /**
- * If this packet is queued, the sender thread will exit
- */
- final QuorumPacket proposalOfDeath = new QuorumPacket();
-
- private LearnerType learnerType = LearnerType.PARTICIPANT;
- public LearnerType getLearnerType() {
- return learnerType;
- }
-
- /**
- * This method will use the thread to send packets added to the
- * queuedPackets list
- *
- * @throws InterruptedException
- */
- private void sendPackets() throws InterruptedException {
- long traceMask = ZooTrace.SERVER_PACKET_TRACE_MASK;
- while (true) {
- try {
- QuorumPacket p;
- p = queuedPackets.poll();
- if (p == null) {
- bufferedOutput.flush();
- p = queuedPackets.take();
- }
-
- if (p == proposalOfDeath) {
- // Packet of death!
- break;
- }
- if (p.getType() == Leader.PING) {
- traceMask = ZooTrace.SERVER_PING_TRACE_MASK;
- }
- if (p.getType() == Leader.PROPOSAL) {
- syncLimitCheck.updateProposal(p.getZxid(), System.nanoTime());
- }
- if (LOG.isTraceEnabled()) {
- ZooTrace.logQuorumPacket(LOG, traceMask, 'o', p);
- }
- oa.writeRecord(p, "packet");
- } catch (IOException e) {
- if (!sock.isClosed()) {
- LOG.warn("Unexpected exception at " + this, e);
- try {
- // this will cause everything to shutdown on
- // this learner handler and will help notify
- // the learner/observer instantaneously
- sock.close();
- } catch(IOException ie) {
- LOG.warn("Error closing socket for handler " + this, ie);
- }
- }
- break;
- }
- }
- }
-
- static public String packetToString(QuorumPacket p) {
- String type;
- String mess = null;
-
- switch (p.getType()) {
- case Leader.ACK:
- type = "ACK";
- break;
- case Leader.COMMIT:
- type = "COMMIT";
- break;
- case Leader.FOLLOWERINFO:
- type = "FOLLOWERINFO";
- break;
- case Leader.NEWLEADER:
- type = "NEWLEADER";
- break;
- case Leader.PING:
- type = "PING";
- break;
- case Leader.PROPOSAL:
- type = "PROPOSAL";
- TxnHeader hdr = new TxnHeader();
- try {
- SerializeUtils.deserializeTxn(p.getData(), hdr);
- // mess = "transaction: " + txn.toString();
- } catch (IOException e) {
- LOG.warn("Unexpected exception",e);
- }
- break;
- case Leader.REQUEST:
- type = "REQUEST";
- break;
- case Leader.REVALIDATE:
- type = "REVALIDATE";
- ByteArrayInputStream bis = new ByteArrayInputStream(p.getData());
- DataInputStream dis = new DataInputStream(bis);
- try {
- long id = dis.readLong();
- mess = " sessionid = " + id;
- } catch (IOException e) {
- LOG.warn("Unexpected exception", e);
- }
-
- break;
- case Leader.UPTODATE:
- type = "UPTODATE";
- break;
- case Leader.DIFF:
- type = "DIFF";
- break;
- case Leader.TRUNC:
- type = "TRUNC";
- break;
- case Leader.SNAP:
- type = "SNAP";
- break;
- case Leader.ACKEPOCH:
- type = "ACKEPOCH";
- break;
- case Leader.SYNC:
- type = "SYNC";
- break;
- case Leader.INFORM:
- type = "INFORM";
- break;
- case Leader.COMMITANDACTIVATE:
- type = "COMMITANDACTIVATE";
- break;
- case Leader.INFORMANDACTIVATE:
- type = "INFORMANDACTIVATE";
- break;
- default:
- type = "UNKNOWN" + p.getType();
- }
- String entry = null;
- if (type != null) {
- entry = type + " " + Long.toHexString(p.getZxid()) + " " + mess;
- }
- return entry;
- }
-
- /**
- * This thread will receive packets from the peer and process them and
- * also listen to new connections from new peers.
- */
- @Override
- public void run() {
- try {
- leader.addLearnerHandler(this);
- tickOfNextAckDeadline = leader.self.tick.get()
- + leader.self.initLimit + leader.self.syncLimit;
-
- ia = BinaryInputArchive.getArchive(bufferedInput);
- bufferedOutput = new BufferedOutputStream(sock.getOutputStream());
- oa = BinaryOutputArchive.getArchive(bufferedOutput);
-
- QuorumPacket qp = new QuorumPacket();
- ia.readRecord(qp, "packet");
- if(qp.getType() != Leader.FOLLOWERINFO && qp.getType() != Leader.OBSERVERINFO){
- LOG.error("First packet " + qp.toString()
- + " is not FOLLOWERINFO or OBSERVERINFO!");
- return;
- }
-
- byte learnerInfoData[] = qp.getData();
- if (learnerInfoData != null) {
- ByteBuffer bbsid = ByteBuffer.wrap(learnerInfoData);
- if (learnerInfoData.length >= 8) {
- this.sid = bbsid.getLong();
- }
- if (learnerInfoData.length >= 12) {
- this.version = bbsid.getInt(); // protocolVersion
- }
- if (learnerInfoData.length >= 20) {
- long configVersion = bbsid.getLong();
- if (configVersion > leader.self.getQuorumVerifier().getVersion()) {
- throw new IOException("Follower is ahead of the leader (has a later activated configuration)");
- }
- }
- } else {
- this.sid = leader.followerCounter.getAndDecrement();
- }
-
- if (leader.self.getView().containsKey(this.sid)) {
- LOG.info("Follower sid: " + this.sid + " : info : "
- + leader.self.getView().get(this.sid).toString());
- } else {
- LOG.info("Follower sid: " + this.sid + " not in the current config " + Long.toHexString(leader.self.getQuorumVerifier().getVersion()));
- }
-
- if (qp.getType() == Leader.OBSERVERINFO) {
- learnerType = LearnerType.OBSERVER;
- }
-
- long lastAcceptedEpoch = ZxidUtils.getEpochFromZxid(qp.getZxid());
-
- long peerLastZxid;
- StateSummary ss = null;
- long zxid = qp.getZxid();
- long newEpoch = leader.getEpochToPropose(this.getSid(), lastAcceptedEpoch);
- long newLeaderZxid = ZxidUtils.makeZxid(newEpoch, 0);
-
- if (this.getVersion() < 0x10000) {
- // we are going to have to extrapolate the epoch information
- long epoch = ZxidUtils.getEpochFromZxid(zxid);
- ss = new StateSummary(epoch, zxid);
- // fake the message
- leader.waitForEpochAck(this.getSid(), ss);
- } else {
- byte ver[] = new byte[4];
- ByteBuffer.wrap(ver).putInt(0x10000);
- QuorumPacket newEpochPacket = new QuorumPacket(Leader.LEADERINFO, newLeaderZxid, ver, null);
- oa.writeRecord(newEpochPacket, "packet");
- bufferedOutput.flush();
- QuorumPacket ackEpochPacket = new QuorumPacket();
- ia.readRecord(ackEpochPacket, "packet");
- if (ackEpochPacket.getType() != Leader.ACKEPOCH) {
- LOG.error(ackEpochPacket.toString()
- + " is not ACKEPOCH");
- return;
- }
- ByteBuffer bbepoch = ByteBuffer.wrap(ackEpochPacket.getData());
- ss = new StateSummary(bbepoch.getInt(), ackEpochPacket.getZxid());
- leader.waitForEpochAck(this.getSid(), ss);
- }
- peerLastZxid = ss.getLastZxid();
-
- // Take any necessary action if we need to send TRUNC or DIFF
- // startForwarding() will be called in all cases
- boolean needSnap = syncFollower(peerLastZxid, leader.zk.getZKDatabase(), leader);
-
- LOG.debug("Sending NEWLEADER message to " + sid);
- // the version of this quorumVerifier will be set by leader.lead() in case
- // the leader is just being established. waitForEpochAck makes sure that readyToStart is true if
- // we got here, so the version was set
- if (getVersion() < 0x10000) {
- QuorumPacket newLeaderQP = new QuorumPacket(Leader.NEWLEADER,
- newLeaderZxid, null, null);
- oa.writeRecord(newLeaderQP, "packet");
- } else {
- QuorumPacket newLeaderQP = new QuorumPacket(Leader.NEWLEADER,
- newLeaderZxid, leader.self.getLastSeenQuorumVerifier()
- .toString().getBytes(), null);
- queuedPackets.add(newLeaderQP);
- }
- bufferedOutput.flush();
-
- /* if we are not truncating or sending a diff just send a snapshot */
- if (needSnap) {
- boolean exemptFromThrottle = getLearnerType() != LearnerType.OBSERVER;
- LearnerSnapshot snapshot =
- leader.getLearnerSnapshotThrottler().beginSnapshot(exemptFromThrottle);
- try {
- long zxidToSend = leader.zk.getZKDatabase().getDataTreeLastProcessedZxid();
- oa.writeRecord(new QuorumPacket(Leader.SNAP, zxidToSend, null, null), "packet");
- bufferedOutput.flush();
-
- LOG.info("Sending snapshot last zxid of peer is 0x{}, zxid of leader is 0x{}, "
- + "send zxid of db as 0x{}, {} concurrent snapshots, "
- + "snapshot was {} from throttle",
- Long.toHexString(peerLastZxid),
- Long.toHexString(leaderLastZxid),
- Long.toHexString(zxidToSend),
- snapshot.getConcurrentSnapshotNumber(),
- snapshot.isEssential() ? "exempt" : "not exempt");
- // Dump data to peer
- leader.zk.getZKDatabase().serializeSnapshot(oa);
- oa.writeString("BenWasHere", "signature");
- bufferedOutput.flush();
- } finally {
- snapshot.close();
- }
- }
-
- // Start thread that blast packets in the queue to learner
- startSendingPackets();
-
- /*
- * Have to wait for the first ACK, wait until
- * the leader is ready, and only then we can
- * start processing messages.
- */
- qp = new QuorumPacket();
- ia.readRecord(qp, "packet");
- if(qp.getType() != Leader.ACK){
- LOG.error("Next packet was supposed to be an ACK,"
- + " but received packet: {}", packetToString(qp));
- return;
- }
-
- if(LOG.isDebugEnabled()){
- LOG.debug("Received NEWLEADER-ACK message from " + sid);
- }
- leader.waitForNewLeaderAck(getSid(), qp.getZxid());
-
- syncLimitCheck.start();
-
- // now that the ack has been processed expect the syncLimit
- sock.setSoTimeout(leader.self.tickTime * leader.self.syncLimit);
-
- /*
- * Wait until leader starts up
- */
- synchronized(leader.zk){
- while(!leader.zk.isRunning() && !this.isInterrupted()){
- leader.zk.wait(20);
- }
- }
- // Mutation packets will be queued during the serialize,
- // so we need to mark when the peer can actually start
- // using the data
- //
- LOG.debug("Sending UPTODATE message to " + sid);
- queuedPackets.add(new QuorumPacket(Leader.UPTODATE, -1, null, null));
-
- while (true) {
- qp = new QuorumPacket();
- ia.readRecord(qp, "packet");
-
- long traceMask = ZooTrace.SERVER_PACKET_TRACE_MASK;
- if (qp.getType() == Leader.PING) {
- traceMask = ZooTrace.SERVER_PING_TRACE_MASK;
- }
- if (LOG.isTraceEnabled()) {
- ZooTrace.logQuorumPacket(LOG, traceMask, 'i', qp);
- }
- tickOfNextAckDeadline = leader.self.tick.get() + leader.self.syncLimit;
-
-
- ByteBuffer bb;
- long sessionId;
- int cxid;
- int type;
-
- switch (qp.getType()) {
- case Leader.ACK:
- if (this.learnerType == LearnerType.OBSERVER) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Received ACK from Observer " + this.sid);
- }
- }
- syncLimitCheck.updateAck(qp.getZxid());
- leader.processAck(this.sid, qp.getZxid(), sock.getLocalSocketAddress());
- break;
- case Leader.PING:
- // Process the touches
- ByteArrayInputStream bis = new ByteArrayInputStream(qp
- .getData());
- DataInputStream dis = new DataInputStream(bis);
- while (dis.available() > 0) {
- long sess = dis.readLong();
- int to = dis.readInt();
- leader.zk.touch(sess, to);
- }
- break;
- case Leader.REVALIDATE:
- bis = new ByteArrayInputStream(qp.getData());
- dis = new DataInputStream(bis);
- long id = dis.readLong();
- int to = dis.readInt();
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- DataOutputStream dos = new DataOutputStream(bos);
- dos.writeLong(id);
- boolean valid = leader.zk.checkIfValidGlobalSession(id, to);
- if (valid) {
- try {
- //set the session owner
- // as the follower that
- // owns the session
- leader.zk.setOwner(id, this);
- } catch (SessionExpiredException e) {
- LOG.error("Somehow session " + Long.toHexString(id) +
- " expired right after being renewed! (impossible)", e);
- }
- }
- if (LOG.isTraceEnabled()) {
- ZooTrace.logTraceMessage(LOG,
- ZooTrace.SESSION_TRACE_MASK,
- "Session 0x" + Long.toHexString(id)
- + " is valid: "+ valid);
- }
- dos.writeBoolean(valid);
- qp.setData(bos.toByteArray());
- queuedPackets.add(qp);
- break;
- case Leader.REQUEST:
- bb = ByteBuffer.wrap(qp.getData());
- sessionId = bb.getLong();
- cxid = bb.getInt();
- type = bb.getInt();
- bb = bb.slice();
- Request si;
- if(type == OpCode.sync){
- si = new LearnerSyncRequest(this, sessionId, cxid, type, bb, qp.getAuthinfo());
- } else {
- si = new Request(null, sessionId, cxid, type, bb, qp.getAuthinfo());
- }
- si.setOwner(this);
- leader.zk.submitLearnerRequest(si);
- break;
- default:
- LOG.warn("unexpected quorum packet, type: {}", packetToString(qp));
- break;
- }
- }
- } catch (IOException e) {
- if (sock != null && !sock.isClosed()) {
- LOG.error("Unexpected exception causing shutdown while sock "
- + "still open", e);
- //close the socket to make sure the
- //other side can see it being close
- try {
- sock.close();
- } catch(IOException ie) {
- // do nothing
- }
- }
- } catch (InterruptedException e) {
- LOG.error("Unexpected exception causing shutdown", e);
- } catch (SnapshotThrottleException e) {
- LOG.error("too many concurrent snapshots: " + e);
- } finally {
- LOG.warn("******* GOODBYE "
- + (sock != null ? sock.getRemoteSocketAddress() : "<null>")
- + " ********");
- shutdown();
- }
- }
-
- /**
- * Start thread that will forward any packet in the queue to the follower
- */
- protected void startSendingPackets() {
- if (!sendingThreadStarted) {
- // Start sending packets
- new Thread() {
- public void run() {
- Thread.currentThread().setName(
- "Sender-" + sock.getRemoteSocketAddress());
- try {
- sendPackets();
- } catch (InterruptedException e) {
- LOG.warn("Unexpected interruption " + e.getMessage());
- }
- }
- }.start();
- sendingThreadStarted = true;
- } else {
- LOG.error("Attempting to start sending thread after it already started");
- }
- }
-
- /**
- * Determine if we need to sync with follower using DIFF/TRUNC/SNAP
- * and setup follower to receive packets from commit processor
- *
- * @param peerLastZxid
- * @param db
- * @param leader
- * @return true if snapshot transfer is needed.
- */
- public boolean syncFollower(long peerLastZxid, ZKDatabase db, Leader leader) {
- /*
- * When leader election is completed, the leader will set its
- * lastProcessedZxid to be (epoch < 32). There will be no txn associated
- * with this zxid.
- *
- * The learner will set its lastProcessedZxid to the same value if
- * it get DIFF or SNAP from the leader. If the same learner come
- * back to sync with leader using this zxid, we will never find this
- * zxid in our history. In this case, we will ignore TRUNC logic and
- * always send DIFF if we have old enough history
- */
- boolean isPeerNewEpochZxid = (peerLastZxid & 0xffffffffL) == 0;
- // Keep track of the latest zxid which already queued
- 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()) {
- /*
- * It is possible that committedLog is empty. In that case
- * setting these value to the latest txn in leader db
- * will reduce the case that we need to handle
- *
- * Here is how each case handle by the if block below
- * 1. lastProcessZxid == peerZxid -> Handle by (2)
- * 2. lastProcessZxid < peerZxid -> Handle by (3)
- * 3. lastProcessZxid > peerZxid -> Handle by (5)
- */
- minCommittedLog = lastProcessedZxid;
- maxCommittedLog = lastProcessedZxid;
- }
-
- /*
- * Here are the cases that we want to handle
- *
- * 1. Force sending snapshot (for testing purpose)
- * 2. Peer and leader is already sync, send empty diff
- * 3. Follower has txn that we haven't seen. This may be old leader
- * so we need to send TRUNC. However, if peer has newEpochZxid,
- * we cannot send TRUNC since the follower has no txnlog
- * 4. Follower is within committedLog range or already in-sync.
- * We may need to send DIFF or TRUNC depending on follower's zxid
- * We always send empty DIFF if follower is already in-sync
- * 5. Follower missed the committedLog. We will try to use on-disk
- * txnlog + committedLog to sync with follower. If that fail,
- * we will send snapshot
- */
-
- if (forceSnapSync) {
- // Force leader to use snapshot to sync with follower
- LOG.warn("Forcing snapshot sync - should not see this in production");
- } else if (lastProcessedZxid == peerLastZxid) {
- // Follower is already sync with us, send empty diff
- 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) {
- // Newer than committedLog, send trunc and done
- 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)) {
- // Follower is within commitLog range
- 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) {
- // Use txnlog and committedLog to sync
-
- // Calculate sizeLimit that we allow to retrieve txnlog from disk
- long sizeLimit = db.calculateTxnLogSizeLimit();
- // This method can return empty iterator if the requested zxid
- // is older than on-disk txnlog
- 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;
- }
- // closing the resources
- 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) {
- // This should never happen, but we should fall back to sending
- // snapshot just in case.
- LOG.error("Unhandled scenario for peer sid: " + getSid() +
- " fall back to use snapshot");
- needSnap = true;
- }
-
- return needSnap;
- }
-
- /**
- * Queue committed proposals into packet queue. The range of packets which
- * is going to be queued are (peerLaxtZxid, maxZxid]
- *
- * @param itr iterator point to the proposals
- * @param peerLastZxid last zxid seen by the follower
- * @param maxZxid max zxid of the proposal to queue, null if no limit
- * @param lastCommittedZxid when sending diff, we need to send lastCommittedZxid
- * on the leader to follow Zab 1.0 protocol.
- * @return last zxid of the queued proposal
- */
- protected long queueCommittedProposals(Iterator<Proposal> itr,
- long peerLastZxid, Long maxZxid, Long lastCommittedZxid) {
- boolean isPeerNewEpochZxid = (peerLastZxid & 0xffffffffL) == 0;
- long queuedZxid = peerLastZxid;
- // as we look through proposals, this variable keeps track of previous
- // proposal Id.
- long prevProposalZxid = -1;
- while (itr.hasNext()) {
- Proposal propose = itr.next();
-
- long packetZxid = propose.packet.getZxid();
- // abort if we hit the limit
- if ((maxZxid != null) && (packetZxid > maxZxid)) {
- break;
- }
-
- // skip the proposals the peer already has
- if (packetZxid < peerLastZxid) {
- prevProposalZxid = packetZxid;
- continue;
- }
-
- // If we are sending the first packet, figure out whether to trunc
- // or diff
- if (needOpPacket) {
-
- // Send diff when we see the follower's zxid in our history
- if (packetZxid == peerLastZxid) {
- LOG.info("Sending DIFF zxid=0x" +
- Long.toHexString(lastCommittedZxid) +
- " for peer sid: " + getSid());
- queueOpPacket(Leader.DIFF, lastCommittedZxid);
- needOpPacket = false;
- continue;
- }
-
- if (isPeerNewEpochZxid) {
- // Send diff and fall through if zxid is of a new-epoch
- LOG.info("Sending DIFF zxid=0x" +
- Long.toHexString(lastCommittedZxid) +
- " for peer sid: " + getSid());
- queueOpPacket(Leader.DIFF, lastCommittedZxid);
- needOpPacket = false;
- } else if (packetZxid > peerLastZxid ) {
- // Peer have some proposals that the leader hasn't seen yet
- // it may used to be a leader
- if (ZxidUtils.getEpochFromZxid(packetZxid) !=
- ZxidUtils.getEpochFromZxid(peerLastZxid)) {
- // We cannot send TRUNC that cross epoch boundary.
- // The learner will crash if it is asked to do so.
- // We will send snapshot this those cases.
- LOG.warn("Cannot send TRUNC to peer sid: " + getSid() +
- " peer zxid is from different epoch" );
- return queuedZxid;
- }
-
- LOG.info("Sending TRUNC zxid=0x" +
- Long.toHexString(prevProposalZxid) +
- " for peer sid: " + getSid());
- queueOpPacket(Leader.TRUNC, prevProposalZxid);
- needOpPacket = false;
- }
- }
-
- if (packetZxid <= queuedZxid) {
- // We can get here, if we don't have op packet to queue
- // or there is a duplicate txn in a given iterator
- continue;
- }
-
- // Since this is already a committed proposal, we need to follow
- // it by a commit packet
- queuePacket(propose.packet);
- queueOpPacket(Leader.COMMIT, packetZxid);
- queuedZxid = packetZxid;
-
- }
-
- if (needOpPacket && isPeerNewEpochZxid) {
- // We will send DIFF for this kind of zxid in any case. This if-block
- // is the catch when our history older than learner and there is
- // no new txn since then. So we need an empty diff
- LOG.info("Sending DIFF zxid=0x" +
- Long.toHexString(lastCommittedZxid) +
- " for peer sid: " + getSid());
- queueOpPacket(Leader.DIFF, lastCommittedZxid);
- needOpPacket = false;
- }
-
- return queuedZxid;
- }
-
- public void shutdown() {
- // Send the packet of death
- try {
- queuedPackets.put(proposalOfDeath);
- } catch (InterruptedException e) {
- LOG.warn("Ignoring unexpected exception", e);
- }
- try {
- if (sock != null && !sock.isClosed()) {
- sock.close();
- }
- } catch (IOException e) {
- LOG.warn("Ignoring unexpected exception during socket close", e);
- }
- this.interrupt();
- leader.removeLearnerHandler(this);
- }
-
- public long tickOfNextAckDeadline() {
- return tickOfNextAckDeadline;
- }
-
- /**
- * ping calls from the leader to the peers
- */
- public void ping() {
- // If learner hasn't sync properly yet, don't send ping packet
- // otherwise, the learner will crash
- if (!sendingThreadStarted) {
- return;
- }
- long id;
- if (syncLimitCheck.check(System.nanoTime())) {
- synchronized(leader) {
- id = leader.lastProposed;
- }
- QuorumPacket ping = new QuorumPacket(Leader.PING, id, null, null);
- queuePacket(ping);
- } else {
- LOG.warn("Closing connection to peer due to transaction timeout.");
- shutdown();
- }
- }
-
- /**
- * Queue leader packet of a given type
- * @param type
- * @param zxid
- */
- private void queueOpPacket(int type, long zxid) {
- QuorumPacket packet = new QuorumPacket(type, zxid, null, null);
- queuePacket(packet);
- }
-
- void queuePacket(QuorumPacket p) {
- queuedPackets.add(p);
- }
-
- public boolean synced() {
- return isAlive()
- && leader.self.tick.get() <= tickOfNextAckDeadline;
- }
-
- /**
- * For testing, return packet queue
- * @return
- */
- public Queue<QuorumPacket> getQueuedPackets() {
- return queuedPackets;
- }
-
- /**
- * For testing, we need to reset this value
- */
- public void setFirstPacket(boolean value) {
- needOpPacket = value;
- }
-}