[ZooKeeper]Client Session失效

前面一篇文章提到zookeeper server端主动发现session超时并清理session信息，关闭连接，接下来看看client端如何试图恢复session的。关于client端代码分析见前文http://iwinit.iteye.com/blog/1754611 。
由于session被清理，此时server端已经没有session信息了。而由于连接被关闭，client会抛出异常

if (sockKey.isReadable()) {
        //返回-1
            int rc = sock.read(incomingBuffer);
            if (rc < 0) {
                throw new EndOfStreamException(
                        "Unable to read additional data from server sessionid 0x"
                                + Long.toHexString(sessionId)
                                + ", likely server has closed socket");
            }

 SendThread处理异常

else {
                       // this is ugly, you have a better way speak up
                       if (e instanceof SessionExpiredException) {
                           LOG.info(e.getMessage() + ", closing socket connection");
                       } else if (e instanceof SessionTimeoutException) {
                           LOG.info(e.getMessage() + RETRY_CONN_MSG);
                       }
        //连接被关闭了
        else if (e instanceof EndOfStreamException) {
                           LOG.info(e.getMessage() + RETRY_CONN_MSG);
                       } else if (e instanceof RWServerFoundException) {
                           LOG.info(e.getMessage());
                       } else {
                           LOG.warn(
                                   "Session 0x"
                                           + Long.toHexString(getSessionId())
                                           + " for server "
                                           + clientCnxnSocket.getRemoteSocketAddress()
                                           + ", unexpected error"
                                           + RETRY_CONN_MSG, e);
                       }
        //清理连接，失败当前请求
                       cleanup();
        //此时state还是CONNECTED,发送DISCONNECTED状态通知
                       if (state.isAlive()) {
                           eventThread.queueEvent(new WatchedEvent(
                                   Event.EventType.None,
                                   Event.KeeperState.Disconnected,
                                   null));
                       }
                       clientCnxnSocket.updateNow();
                       clientCnxnSocket.updateLastSendAndHeard();
                   }

 接下来client重新寻找下一个server进行session恢复，此时client的sessionId和password仍然是上一次创建的session信息。

//此处返回true，因为连接已经被置为null了
            if (!clientCnxnSocket.isConnected()) {
            //重连时，先sleep一下
                        if(!isFirstConnect){
                            try {
                                Thread.sleep(r.nextInt(1000));
                            } catch (InterruptedException e) {
                                LOG.warn("Unexpected exception", e);
                            }
                        }
                        // don't re-establish connection if we are closing
                        if (closing || !state.isAlive()) {
                            break;
                        }
            //重新开始连接
                        startConnect();
                        clientCnxnSocket.updateLastSendAndHeard();
                    }

 取下一个server

addr = hostProvider.next(1000);

 之后就和新建session时类似，区别是发送ConnectRequest时sessionid和password都是老的

//新建session成功时，seenRwServerBefore已经被置为true
long sessId = (seenRwServerBefore) ? sessionId : 0;
            ConnectRequest conReq = new ConnectRequest(0, lastZxid,
                    sessionTimeout, sessId, sessionPasswd);

 接下来server端处理

//此时sessionId不为0
 long sessionId = connReq.getSessionId();
        if (sessionId != 0) {
            long clientSessionId = connReq.getSessionId();
            LOG.info("Client attempting to renew session 0x"
                    + Long.toHexString(clientSessionId)
                    + " at " + cnxn.getRemoteSocketAddress());
        //先关闭老的连接，如果有的话，删除watch
            serverCnxnFactory.closeSession(sessionId);
            cnxn.setSessionId(sessionId);
            reopenSession(cnxn, sessionId, passwd, sessionTimeout);
        } else {
            LOG.info("Client attempting to establish new session at "
                    + cnxn.getRemoteSocketAddress());
            createSession(cnxn, passwd, sessionTimeout);
        }

 重启session

   public void reopenSession(ServerCnxn cnxn, long sessionId, byte[] passwd,
           int sessionTimeout) throws IOException {
//检查密码，如果不一样，则结束session，返回client一个为0的sessionid。如果sessionid为0，则为false
       if (!checkPasswd(sessionId, passwd)) {
           finishSessionInit(cnxn, false);
       }
//密码正确，再校验下session是否还有效，这里不同的server处理不一样
else {
           revalidateSession(cnxn, sessionId, sessionTimeout);
       }
   }

 重试的server如果是leader

@Override
protected void revalidateSession(ServerCnxn cnxn, long sessionId,
    int sessionTimeout) throws IOException {
类中通过sessionTrack检查
    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.
    }
}

 父类revalidateSession方法

protected void revalidateSession(ServerCnxn cnxn, long sessionId,
        int sessionTimeout) throws IOException {
    boolean rc = sessionTracker.touchSession(sessionId, sessionTimeout);
    if (LOG.isTraceEnabled()) {
        ZooTrace.logTraceMessage(LOG,ZooTrace.SESSION_TRACE_MASK,
                                 "Session 0x" + Long.toHexString(sessionId) +
                " is valid: " + rc);
    }
    finishSessionInit(cnxn, rc);
}

 leader的SessionTracker为SessionTrackerImpl，touchSession方法如下

synchronized public boolean touchSession(long sessionId, int timeout) {
        if (LOG.isTraceEnabled()) {
            ZooTrace.logTraceMessage(LOG,
                                     ZooTrace.CLIENT_PING_TRACE_MASK,
                                     "SessionTrackerImpl --- Touch session: 0x"
                    + Long.toHexString(sessionId) + " with timeout " + timeout);
        }
    //因为session超时，session已经被删掉了，此处返回null，所以检查结果是false
        SessionImpl s = sessionsById.get(sessionId);
        // Return false, if the session doesn't exists or marked as closing
        if (s == null || s.isClosing()) {
            return false;
        }
.....

 对于leader来说session检查的结果是false。

如果是follower，其校验方法

protected void revalidateSession(ServerCnxn cnxn, long sessionId,
            int sessionTimeout) throws IOException {
    //需要询问leader，session是否还有效
        getLearner().validateSession(cnxn, sessionId, sessionTimeout);
    }

 follower询问session是否有效

 */
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();
EVALIDATE包用来检查session是否还有效
    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);
}

 leader端处理REVALIDATE包

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);
            //这里由于session已经被删掉，返回false
                    boolean valid = leader.zk.touch(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);
                    }
            //结果是false
                    dos.writeBoolean(valid);
                    qp.setData(bos.toByteArray());
                    queuedPackets.add(qp);
                    break;

 follower处理返回结果

case Leader.REVALIDATE:
            revalidate(qp);
            break;
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);
        }
    }

 可以看到无论是leader还是follower最后都会调用zk.finishSessionInit(cnxn, valid)处理，而由于session已经失效，所以valid为false

public void finishSessionInit(ServerCnxn cnxn, boolean valid) {
        ......

        try {
        //由于valid是false，所以返回给client的sessionid为0，password为空
            ConnectResponse rsp = new ConnectResponse(0, valid ? cnxn.getSessionTimeout()
                    : 0, valid ? cnxn.getSessionId() : 0, // send 0 if session is no
                            // longer valid
                            valid ? generatePasswd(cnxn.getSessionId()) : new byte[16]);
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            BinaryOutputArchive bos = BinaryOutputArchive.getArchive(baos);
            bos.writeInt(-1, "len");
            rsp.serialize(bos, "connect");
            if (!cnxn.isOldClient) {
                bos.writeBool(
                        this instanceof ReadOnlyZooKeeperServer, "readOnly");
            }
            baos.close();
            ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
            bb.putInt(bb.remaining() - 4).rewind();
            cnxn.sendBuffer(bb);

            ......
    }

 client端处理ConnectResponse

void readConnectResult() throws IOException {
        ......
    //被server重置为0了
        this.sessionId = conRsp.getSessionId();
        sendThread.onConnected(conRsp.getTimeOut(), this.sessionId,
                conRsp.getPasswd(), isRO);
    }

 

void onConnected(int _negotiatedSessionTimeout, long _sessionId,
                byte[] _sessionPasswd, boolean isRO) throws IOException {
            negotiatedSessionTimeout = _negotiatedSessionTimeout;
        //sessionid为0，抛出SessionExpiredException异常
            if (negotiatedSessionTimeout <= 0) {
        //state设为CLOSED，这个行为将关闭SendThread
                state = States.CLOSED;
        //Expired状态通知
                eventThread.queueEvent(new WatchedEvent(
                        Watcher.Event.EventType.None,
                        Watcher.Event.KeeperState.Expired, null));
        //这个行为将关闭EventThread
                eventThread.queueEventOfDeath();
        //抛出异常
                throw new SessionExpiredException(
                        "Unable to reconnect to ZooKeeper service, session 0x"
                                + Long.toHexString(sessionId) + " has expired");
            }
           ......
        }

 SendThread处理SessionExpiredException，关闭SendThread

while (state.isAlive()) {
        ......
        // this is ugly, you have a better way speak up
                       if (e instanceof SessionExpiredException) {
                           LOG.info(e.getMessage() + ", closing socket connection");
                       }
        ......
        //关闭连接，失败所有请求
                       cleanup();
        //此时state已经被置为CLOSED，SendThread将退出
                       if (state.isAlive()) {
                           eventThread.queueEvent(new WatchedEvent(
                                   Event.EventType.None,
                                   Event.KeeperState.Disconnected,
                                   null));
                       }
                       clientCnxnSocket.updateNow();
                       clientCnxnSocket.updateLastSendAndHeard();
    }
    //清理
    cleanup();
    //关闭selector
            clientCnxnSocket.close();
        if (state.isAlive()) {
            eventThread.queueEvent(new WatchedEvent(Event.EventType.None,
                    Event.KeeperState.Disconnected, null));
        }
        ZooTrace.logTraceMessage(LOG, ZooTrace.getTextTraceLevel(),
                                 "SendThread exitedloop.");

 清理动作

private void cleanup() {
        //关闭socket
            clientCnxnSocket.cleanup();
        //发送完等待响应的请求失败，此时由于state是CLOSED，所以异常信息是SESSIONEXPIRED
            synchronized (pendingQueue) {
                for (Packet p : pendingQueue) {
                    conLossPacket(p);
                }
                pendingQueue.clear();
            }
        //等待发送的请求失败，此时由于state是CLOSED，所以异常信息是SESSIONEXPIRED
            synchronized (outgoingQueue) {
                for (Packet p : outgoingQueue) {
                    conLossPacket(p);
                }
                outgoingQueue.clear();
            }
        }

 

private void conLossPacket(Packet p) {
        if (p.replyHeader == null) {
            return;
        }
        switch (state) {
        case AUTH_FAILED:
            p.replyHeader.setErr(KeeperException.Code.AUTHFAILED.intValue());
            break;
    //关闭的时候，是SESSIONEXPIRED异常码
        case CLOSED:
            p.replyHeader.setErr(KeeperException.Code.SESSIONEXPIRED.intValue());
            break;
    //其他是CONNECTIONLOSS异常码
        default:
            p.replyHeader.setErr(KeeperException.Code.CONNECTIONLOSS.intValue());
        }
        finishPacket(p);
    }

 EventThread关闭

public void run() {
           try {
              isRunning = true;
              while (true) {
                 Object event = waitingEvents.take();
        //kill signal
                 if (event == eventOfDeath) {
                    wasKilled = true;
                 } else {
                    processEvent(event);
                 }
                 if (wasKilled)
            //等所有通知都发完再退出
                    synchronized (waitingEvents) {
                       if (waitingEvents.isEmpty()) {
                          isRunning = false;
                          break;
                       }
                    }
              }
           } catch (InterruptedException e) {
              LOG.error("Event thread exiting due to interruption", e);
           }

            LOG.info("EventThread shut down");
        }

 之后client将不可用，所有请求都将发送的时候都将收到SESSIONEXPIRED异常码，因为queuePacket的时候一个判断

Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,
            Record response, AsyncCallback cb, String clientPath,
            String serverPath, Object ctx, WatchRegistration watchRegistration)
    {
        Packet packet = null;

        // Note that we do not generate the Xid for the packet yet. It is
        // generated later at send-time, by an implementation of ClientCnxnSocket::doIO(),
        // where the packet is actually sent.
        synchronized (outgoingQueue) {
            packet = new Packet(h, r, request, response, watchRegistration);
            packet.cb = cb;
            packet.ctx = ctx;
            packet.clientPath = clientPath;
            packet.serverPath = serverPath;
        //此时状态为CLOSED
            if (!state.isAlive() || closing) {
                conLossPacket(packet);
            } else {
                // If the client is asking to close the session then
                // mark as closing
                if (h.getType() == OpCode.closeSession) {
                    closing = true;
                }
                outgoingQueue.add(packet);
            }
        }
        sendThread.getClientCnxnSocket().wakeupCnxn();
        return packet;
    }

 从以上分析可知，SESSIONEXPIRED异常码是比较严重的事件，之后这个zookeeper实例不可用了，如果需要恢复，则需要重新创建zookeeper实例。而CONNECTIONLOSS异常码是比较常见的，比如网络暂时中断的时候，这个状态码下zookeeper会自动重连恢复，因为server端还保留着session信息。