diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java index 3713646d631..3859382fadd 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java @@ -116,6 +116,16 @@ public class ClientCnxn { */ private static final int SET_WATCHES_MAX_LENGTH = 128 * 1024; + /* predefined xid's values recognized as special by the server */ + // -1 means notification(WATCHER_EVENT) + public static final int NOTIFICATION_XID = -1; + // -2 is the xid for pings + public static final int PING_XID = -2; + // -4 is the xid for AuthPacket + public static final int AUTHPACKET_XID = -4; + // -8 is the xid for setWatch + public static final int SET_WATCHES_XID = -8; + static class AuthData { AuthData(String scheme, byte[] data) { @@ -857,16 +867,14 @@ void readResponse(ByteBuffer incomingBuffer) throws IOException { ReplyHeader replyHdr = new ReplyHeader(); replyHdr.deserialize(bbia, "header"); - if (replyHdr.getXid() == -2) { - // -2 is the xid for pings + if (replyHdr.getXid() == PING_XID) { LOG.debug( "Got ping response for session id: 0x{} after {}ms.", Long.toHexString(sessionId), ((System.nanoTime() - lastPingSentNs) / 1000000)); return; } - if (replyHdr.getXid() == -4) { - // -4 is the xid for AuthPacket + if (replyHdr.getXid() == AUTHPACKET_XID) { if (replyHdr.getErr() == KeeperException.Code.AUTHFAILED.intValue()) { state = States.AUTH_FAILED; eventThread.queueEvent(new WatchedEvent(Watcher.Event.EventType.None, Watcher.Event.KeeperState.AuthFailed, null)); @@ -875,8 +883,7 @@ void readResponse(ByteBuffer incomingBuffer) throws IOException { LOG.debug("Got auth session id: 0x{}", Long.toHexString(sessionId)); return; } - if (replyHdr.getXid() == -1) { - // -1 means notification + if (replyHdr.getXid() == NOTIFICATION_XID) { LOG.debug("Got notification session id: 0x{}", Long.toHexString(sessionId)); WatcherEvent event = new WatcherEvent(); event.deserialize(bbia, "response"); @@ -1048,7 +1055,7 @@ record = new SetWatches2(setWatchesLastZxid, dataWatchesBatch, existWatchesBatch childWatchesBatch, persistentWatchesBatch, persistentRecursiveWatchesBatch); opcode = OpCode.setWatches2; } - RequestHeader header = new RequestHeader(-8, opcode); + RequestHeader header = new RequestHeader(ClientCnxn.SET_WATCHES_XID, opcode); Packet packet = new Packet(header, new ReplyHeader(), record, null, null); outgoingQueue.addFirst(packet); } @@ -1058,7 +1065,7 @@ record = new SetWatches2(setWatchesLastZxid, dataWatchesBatch, existWatchesBatch for (AuthData id : authInfo) { outgoingQueue.addFirst( new Packet( - new RequestHeader(-4, OpCode.auth), + new RequestHeader(ClientCnxn.AUTHPACKET_XID, OpCode.auth), null, new AuthPacket(0, id.scheme, id.data), null, @@ -1088,7 +1095,7 @@ private List prependChroot(List paths) { private void sendPing() { lastPingSentNs = System.nanoTime(); - RequestHeader h = new RequestHeader(-2, OpCode.ping); + RequestHeader h = new RequestHeader(ClientCnxn.PING_XID, OpCode.ping); queuePacket(h, null, null, null, null, null, null, null, null); } @@ -1657,7 +1664,7 @@ public void addAuthInfo(String scheme, byte[] auth) { } authInfo.add(new AuthData(scheme, auth)); queuePacket( - new RequestHeader(-4, OpCode.auth), + new RequestHeader(ClientCnxn.AUTHPACKET_XID, OpCode.auth), null, new AuthPacket(0, scheme, auth), null, diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java index 0ffa5af25ea..da8f1143a8a 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java @@ -27,6 +27,7 @@ import java.util.Set; import org.apache.commons.lang.StringUtils; import org.apache.jute.Record; +import org.apache.zookeeper.ClientCnxn; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.KeeperException.Code; import org.apache.zookeeper.KeeperException.SessionMovedException; @@ -200,7 +201,7 @@ public void processRequest(Request request) { lastOp = "PING"; updateStats(request, lastOp, lastZxid); - cnxn.sendResponse(new ReplyHeader(-2, lastZxid, 0), null, "response"); + cnxn.sendResponse(new ReplyHeader(ClientCnxn.PING_XID, lastZxid, 0), null, "response"); return; } case OpCode.createSession: { diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/NIOServerCnxn.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/NIOServerCnxn.java index 91571747347..a3160101ea7 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/NIOServerCnxn.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/NIOServerCnxn.java @@ -34,6 +34,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import org.apache.jute.BinaryInputArchive; import org.apache.jute.Record; +import org.apache.zookeeper.ClientCnxn; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.data.Id; import org.apache.zookeeper.data.Stat; @@ -701,7 +702,7 @@ public void sendResponse(ReplyHeader h, Record r, String tag, String cacheKey, S */ @Override public void process(WatchedEvent event) { - ReplyHeader h = new ReplyHeader(-1, -1L, 0); + ReplyHeader h = new ReplyHeader(ClientCnxn.NOTIFICATION_XID, -1L, 0); if (LOG.isTraceEnabled()) { ZooTrace.logTraceMessage( LOG, diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxn.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxn.java index ca89672144d..f51fd013706 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxn.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxn.java @@ -40,6 +40,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import org.apache.jute.BinaryInputArchive; import org.apache.jute.Record; +import org.apache.zookeeper.ClientCnxn; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.data.Id; import org.apache.zookeeper.data.Stat; @@ -147,7 +148,7 @@ public int getSessionTimeout() { @Override public void process(WatchedEvent event) { - ReplyHeader h = new ReplyHeader(-1, -1L, 0); + ReplyHeader h = new ReplyHeader(ClientCnxn.NOTIFICATION_XID, -1L, 0); if (LOG.isTraceEnabled()) { ZooTrace.logTraceMessage( LOG, diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/WatchLeakTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/WatchLeakTest.java index cdc3b830470..2455738e4b8 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/WatchLeakTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/WatchLeakTest.java @@ -43,6 +43,7 @@ import java.util.Random; import org.apache.jute.InputArchive; import org.apache.jute.OutputArchive; +import org.apache.zookeeper.ClientCnxn; import org.apache.zookeeper.MockPacket; import org.apache.zookeeper.ZKParameterized; import org.apache.zookeeper.ZooDefs; @@ -251,7 +252,7 @@ private ByteBuffer createWatchesMessage() { SetWatches sw = new SetWatches(1L, dataWatches, existWatches, childWatches); RequestHeader h = new RequestHeader(); h.setType(ZooDefs.OpCode.setWatches); - h.setXid(-8); + h.setXid(ClientCnxn.SET_WATCHES_XID); MockPacket p = new MockPacket(h, new ReplyHeader(), sw, null, null); return p.createAndReturnBB(); }