ZOOKEEPER-2939: Added last/min/max proposal size JMX beans

https://issues.apache.org/jira/browse/ZOOKEEPER-2939
- Refactor proposal serialization logic to a common place (SerializeUtils),
- Add JMX metric to monitor jute.maxbuffer setting
- Add JMX metric to monitor min/max/last size of generated proposals
- Add proposal stats info to 'stat' 4lw command
- Add proposal stats resetting to 'srst' 4lw command
- Unit tests

Author: Andor Molnár <andor@cloudera.com>
Author: Andor Molnar <andor@cloudera.com>

Reviewers: Patrick Hunt <phunt@apache.org>, Abraham Fine <afine@apache.org>

Closes #415 from anmolnar/ZOOKEEPER-2933 and squashes the following commits:

cf6ea5ecb [Andor Molnar] ZOOKEEPER-2933. Added proposal size statistics to 'mntr' command
cb28df6d6 [Andor Molnár] ZOOKEEPER-2939. Fixed findbugs issue: newline in format string
8e6b65f9f [Andor Molnár] ZOOKEEPER-2939: Added last/min/max proposal size JMX beans

build.xml

@@ -1826,7 +1826,7 @@ xmlns:cs="antlib:com.puppycrawl.tools.checkstyle.ant">
              description="Create eclipse project files">
        <ivy:resolve useOrigin="true" conf="test"/>
        <ivy:cachepath pathid="default.path.id" conf="default" />
-       <ivy:cachepath pathid="junit.path.id" conf="test" transitive="false"/>
+       <ivy:cachepath pathid="junit.path.id" conf="test" />
        <taskdef name="eclipse"
                 classname="prantl.ant.eclipse.EclipseTask"
                 classpathref="java.classpath" />

src/java/main/org/apache/zookeeper/server/ZKDatabase.java

@@ -18,7 +18,6 @@
 
 package org.apache.zookeeper.server;
 
-import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.PrintWriter;
 import java.util.Collection;
@@ -32,7 +31,6 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
 
-import org.apache.jute.BinaryOutputArchive;
 import org.apache.jute.InputArchive;
 import org.apache.jute.OutputArchive;
 import org.apache.jute.Record;
@@ -264,19 +262,8 @@ public class ZKDatabase {
                 maxCommittedLog = request.zxid;
             }
 
-            ByteArrayOutputStream baos = new ByteArrayOutputStream();
-            BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
-            try {
-                request.getHdr().serialize(boa, "hdr");
-                if (request.getTxn() != null) {
-                    request.getTxn().serialize(boa, "txn");
-                }
-                baos.close();
-            } catch (IOException e) {
-                LOG.error("This really should be impossible", e);
-            }
-            QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid,
-                    baos.toByteArray(), null);
+            byte[] data = SerializeUtils.serializeRequest(request);
+            QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid, data, null);
             Proposal p = new Proposal();
             p.packet = pp;
             p.request = request;

src/java/main/org/apache/zookeeper/server/ZooKeeperServerBean.java

@@ -20,6 +20,7 @@ package org.apache.zookeeper.server;
 
 import java.util.Date;
 
+import org.apache.jute.BinaryInputArchive;
 import org.apache.zookeeper.Version;
 import org.apache.zookeeper.jmx.ZKMBeanInfo;
 
@@ -167,4 +168,9 @@ public class ZooKeeperServerBean implements ZooKeeperServerMXBean, ZKMBeanInfo {
     public long getTxnLogElapsedSyncTime() {
         return zks.getTxnLogElapsedSyncTime();
     }
+
+    @Override
+    public int getJuteMaxBufferSize() {
+        return BinaryInputArchive.maxBuffer;
+    }
 }

src/java/main/org/apache/zookeeper/server/ZooKeeperServerMXBean.java

@@ -130,4 +130,9 @@ public interface ZooKeeperServerMXBean {
      * Returns the elapsed sync of time of transaction log in milliseconds.
      */
     public long getTxnLogElapsedSyncTime();
+
+    /**
+     * @return Returns the value of the following config setting: jute.maxbuffer
+     */
+    public int getJuteMaxBufferSize();
 }

src/java/main/org/apache/zookeeper/server/command/AbstractFourLetterCommand.java

@@ -35,7 +35,9 @@ import org.slf4j.LoggerFactory;
 public abstract class AbstractFourLetterCommand {
     private static final Logger LOG = LoggerFactory
         .getLogger(AbstractFourLetterCommand.class);
-    protected static final String ZK_NOT_SERVING = "This ZooKeeper instance is not currently serving requests";
+
+    public static final String ZK_NOT_SERVING = "This ZooKeeper instance is not currently serving requests";
+
     protected PrintWriter pw;
     protected ServerCnxn serverCnxn;
     protected ZooKeeperServer zkServer;

src/java/main/org/apache/zookeeper/server/command/MonitorCommand.java

@@ -74,6 +74,10 @@ public class MonitorCommand extends AbstractFourLetterCommand {
             print("followers", leader.getLearners().size());
             print("synced_followers", leader.getForwardingFollowers().size());
             print("pending_syncs", leader.getNumPendingSyncs());
+
+            print("last_proposal_size", leader.getProposalStats().getLastProposalSize());
+            print("max_proposal_size", leader.getProposalStats().getMaxProposalSize());
+            print("min_proposal_size", leader.getProposalStats().getMinProposalSize());
         }
     }
 

src/java/main/org/apache/zookeeper/server/command/StatCommand.java

@@ -22,6 +22,10 @@ import java.io.PrintWriter;
 
 import org.apache.zookeeper.Version;
 import org.apache.zookeeper.server.ServerCnxn;
+import org.apache.zookeeper.server.ServerStats;
+import org.apache.zookeeper.server.quorum.Leader;
+import org.apache.zookeeper.server.quorum.LeaderZooKeeperServer;
+import org.apache.zookeeper.server.quorum.ProposalStats;
 import org.apache.zookeeper.server.quorum.ReadOnlyZooKeeperServer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -54,9 +58,15 @@ public class StatCommand extends AbstractFourLetterCommand {
                 }
                 pw.println();
             }
-            pw.print(zkServer.serverStats().toString());
+            ServerStats serverStats = zkServer.serverStats();
+            pw.print(serverStats.toString());
             pw.print("Node count: ");
             pw.println(zkServer.getZKDatabase().getNodeCount());
+            if (serverStats.getServerState().equals("leader")) {
+                Leader leader = ((LeaderZooKeeperServer)zkServer).getLeader();
+                ProposalStats proposalStats = leader.getProposalStats();
+                pw.printf("Proposal sizes last/min/max: %s%n", proposalStats.toString());
+            }
         }
     }
 }

src/java/main/org/apache/zookeeper/server/command/StatResetCommand.java

@@ -21,6 +21,8 @@ package org.apache.zookeeper.server.command;
 import java.io.PrintWriter;
 
 import org.apache.zookeeper.server.ServerCnxn;
+import org.apache.zookeeper.server.ServerStats;
+import org.apache.zookeeper.server.quorum.LeaderZooKeeperServer;
 
 public class StatResetCommand extends AbstractFourLetterCommand {
     public StatResetCommand(PrintWriter pw, ServerCnxn serverCnxn) {
@@ -32,7 +34,11 @@ public class StatResetCommand extends AbstractFourLetterCommand {
         if (!isZKServerRunning()) {
             pw.println(ZK_NOT_SERVING);
         } else {
-            zkServer.serverStats().reset();
+            ServerStats serverStats = zkServer.serverStats();
+            serverStats.reset();
+            if (serverStats.getServerState().equals("leader")) {
+                ((LeaderZooKeeperServer)zkServer).getLeader().getProposalStats().reset();
+            }
             pw.println("Server stats reset.");
         }
     }

src/java/main/org/apache/zookeeper/server/quorum/Leader.java

@@ -51,6 +51,7 @@ import org.apache.zookeeper.server.RequestProcessor;
 import org.apache.zookeeper.server.ZooKeeperCriticalThread;
 import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
 import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
+import org.apache.zookeeper.server.util.SerializeUtils;
 import org.apache.zookeeper.server.util.ZxidUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -105,6 +106,12 @@ public class Leader {
     private final HashSet<LearnerHandler> learners =
         new HashSet<LearnerHandler>();
 
+    private final ProposalStats proposalStats;
+
+    public ProposalStats getProposalStats() {
+        return proposalStats;
+    }
+
     /**
      * Returns a copy of the current learner snapshot
      */
@@ -222,6 +229,7 @@ public class Leader {
 
     Leader(QuorumPeer self,LeaderZooKeeperServer zk) throws IOException {
         this.self = self;
+        this.proposalStats = new ProposalStats();
         try {
             if (self.getQuorumListenOnAllIPs()) {
                 ss = new ServerSocket(self.getQuorumAddress().getPort());
@@ -721,7 +729,6 @@ public class Leader {
 
     /**
      * @return True if committed, otherwise false.
-     * @param a proposal p
      **/
     synchronized public boolean tryToCommit(Proposal p, long zxid, SocketAddress followerAddr) {       
        // make sure that ops are committed in order. With reconfigurations it is now possible
@@ -992,8 +999,6 @@ public class Leader {
 
     /**
      * Create an inform packet and send it to all observers.
-     * @param zxid
-     * @param proposal
      */
     public void inform(Proposal proposal) {
         QuorumPacket qp = new QuorumPacket(Leader.INFORM, proposal.request.zxid,
@@ -1004,8 +1009,6 @@ public class Leader {
     
     /**
      * Create an inform&activate packet and send it to all observers.
-     * @param zxid
-     * @param proposal
      */
     public void informAndActivate(Proposal proposal, long designatedLeader) {
        byte[] proposalData = proposal.packet.getData();
@@ -1055,19 +1058,9 @@ public class Leader {
             throw new XidRolloverException(msg);
         }
 
-        ByteArrayOutputStream baos = new ByteArrayOutputStream();
-        BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
-        try {
-            request.getHdr().serialize(boa, "hdr");
-            if (request.getTxn() != null) {
-                request.getTxn().serialize(boa, "txn");
-            }
-            baos.close();
-        } catch (IOException e) {
-            LOG.warn("This really should be impossible", e);
-        }
-        QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid,
-                baos.toByteArray(), null);
+        byte[] data = SerializeUtils.serializeRequest(request);
+        proposalStats.setLastProposalSize(data.length);
+        QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid, data, null);
 
         Proposal p = new Proposal();
         p.packet = pp;
@@ -1120,11 +1113,7 @@ public class Leader {
 
     /**
      * Sends a sync message to the appropriate server
-     *
-     * @param f
-     * @param r
      */
-
     public void sendSync(LearnerSyncRequest r){
         QuorumPacket qp = new QuorumPacket(Leader.SYNC, 0, null, null);
         r.fh.queuePacket(qp);

src/java/main/org/apache/zookeeper/server/quorum/LeaderBean.java

@@ -20,8 +20,6 @@ package org.apache.zookeeper.server.quorum;
 
 import org.apache.zookeeper.server.ZooKeeperServerBean;
 import org.apache.zookeeper.server.ZooKeeperServer;
-import org.apache.zookeeper.server.quorum.LearnerHandler;
-import org.apache.zookeeper.server.quorum.Leader;
 
 /**
  * Leader MBean interface implementation.
@@ -54,4 +52,24 @@ public class LeaderBean extends ZooKeeperServerBean implements LeaderMXBean {
     public long getElectionTimeTaken() {
         return leader.self.getElectionTimeTaken();
     }
+
+    @Override
+    public int getLastProposalSize() {
+        return leader.getProposalStats().getLastProposalSize();
+    }
+
+    @Override
+    public int getMinProposalSize() {
+        return leader.getProposalStats().getMinProposalSize();
+    }
+
+    @Override
+    public int getMaxProposalSize() {
+        return leader.getProposalStats().getMaxProposalSize();
+    }
+
+    @Override
+    public void resetProposalStatistics() {
+        leader.getProposalStats().reset();
+    }
 }

src/java/main/org/apache/zookeeper/server/quorum/LeaderMXBean.java

@@ -38,4 +38,24 @@ public interface LeaderMXBean extends ZooKeeperServerMXBean {
      * @return time taken for leader election in milliseconds.
      */
     public long getElectionTimeTaken();
+
+    /**
+     * @return size of latest generated proposal
+     */
+    public int getLastProposalSize();
+
+    /**
+     * @return size of smallest generated proposal
+     */
+    public int getMinProposalSize();
+
+    /**
+     * @return size of largest generated proposal
+     */
+    public int getMaxProposalSize();
+
+    /**
+     * Resets statistics of proposal size (min/max/last)
+     */
+    public void resetProposalStatistics();
 }

src/java/main/org/apache/zookeeper/server/quorum/ProposalStats.java

@@ -0,0 +1,71 @@
+/**
+ * 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;
+
+/**
+ * Provides live statistics about a running Leader.
+ */
+public class ProposalStats {
+    /**
+     * Size of the last generated proposal. This should fit into server's jute.maxbuffer setting.
+     */
+    private int lastProposalSize = -1;
+
+    /**
+     * Size of the smallest proposal which has been generated since the server was started.
+     */
+    private int minProposalSize = -1;
+
+    /**
+     * Size of the largest proposal which has been generated since the server was started.
+     */
+    private int maxProposalSize = -1;
+
+    public synchronized int getLastProposalSize() {
+        return lastProposalSize;
+    }
+
+    synchronized void setLastProposalSize(int value) {
+        lastProposalSize = value;
+        if (minProposalSize == -1 || value < minProposalSize) {
+            minProposalSize = value;
+        }
+        if (value > maxProposalSize) {
+            maxProposalSize = value;
+        }
+    }
+
+    public synchronized int getMinProposalSize() {
+        return minProposalSize;
+    }
+
+    public synchronized int getMaxProposalSize() {
+        return maxProposalSize;
+    }
+
+    public synchronized void reset() {
+        lastProposalSize = -1;
+        minProposalSize = -1;
+        maxProposalSize = -1;
+    }
+
+    public synchronized String toString() {
+        return String.format("%d/%d/%d", lastProposalSize, minProposalSize, maxProposalSize);
+    }
+}

src/java/main/org/apache/zookeeper/server/util/SerializeUtils.java

@@ -19,11 +19,14 @@
 package org.apache.zookeeper.server.util;
 
 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.ZooDefs.OpCode;
+import org.apache.zookeeper.common.IOUtils;
 import org.apache.zookeeper.server.DataTree;
+import org.apache.zookeeper.server.Request;
 import org.apache.zookeeper.server.ZooTrace;
 import org.apache.zookeeper.txn.CreateContainerTxn;
 import org.apache.zookeeper.txn.CreateSessionTxn;
@@ -40,6 +43,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.EOFException;
 import java.io.IOException;
 import java.util.HashMap;
@@ -148,4 +152,20 @@ public class SerializeUtils {
         dt.serialize(oa, "tree");
     }
 
+    public static byte[] serializeRequest(Request request) {
+        if (request == null || request.getHdr() == null) return null;
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
+        try {
+            request.getHdr().serialize(boa, "hdr");
+            if (request.getTxn() != null) {
+                request.getTxn().serialize(boa, "txn");
+            }
+        } catch (IOException e) {
+            LOG.error("This really should be impossible", e);
+        } finally {
+            IOUtils.cleanup(LOG, baos);
+        }
+        return baos.toByteArray();
+    }
 }

src/java/test/org/apache/zookeeper/server/ZooKeeperServerStartupTest.java

@@ -18,6 +18,7 @@
 package org.apache.zookeeper.server;
 
 import static org.apache.zookeeper.client.FourLetterWordMain.send4LetterWord;
+import static org.apache.zookeeper.server.command.AbstractFourLetterCommand.ZK_NOT_SERVING;
 
 import java.io.File;
 import java.io.IOException;
@@ -45,7 +46,6 @@ public class ZooKeeperServerStartupTest extends ZKTestCase {
     private static int PORT = PortAssignment.unique();
     private static String HOST = "127.0.0.1";
     private static String HOSTPORT = HOST + ":" + PORT;
-    private static final String ZK_NOT_SERVING = "This ZooKeeper instance is not currently serving requests";
 
     private ServerCnxnFactory servcnxnf;
     private ZooKeeperServer zks;

src/java/test/org/apache/zookeeper/server/quorum/LeaderBeanTest.java

@@ -0,0 +1,151 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.jute.OutputArchive;
+import org.apache.jute.Record;
+import org.apache.zookeeper.server.Request;
+import org.apache.zookeeper.server.ZKDatabase;
+import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
+import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
+import org.apache.zookeeper.server.util.SerializeUtils;
+import org.apache.zookeeper.test.ClientBase;
+import org.apache.zookeeper.txn.TxnHeader;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import java.io.File;
+import java.io.IOException;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+
+public class LeaderBeanTest {
+    private Leader leader;
+    private LeaderBean leaderBean;
+    private FileTxnSnapLog fileTxnSnapLog;
+    private LeaderZooKeeperServer zks;
+    private QuorumPeer qp;
+
+    @Before
+    public void setUp() throws IOException {
+        qp = new QuorumPeer();
+        QuorumVerifier quorumVerifierMock = mock(QuorumVerifier.class);
+        qp.setQuorumVerifier(quorumVerifierMock, false);
+        File tmpDir = ClientBase.createEmptyTestDir();
+        fileTxnSnapLog = new FileTxnSnapLog(new File(tmpDir, "data"),
+                new File(tmpDir, "data_txnlog"));
+        ZKDatabase zkDb = new ZKDatabase(fileTxnSnapLog);
+
+        zks = new LeaderZooKeeperServer(fileTxnSnapLog, qp, zkDb);
+        leader = new Leader(qp, zks);
+        leaderBean = new LeaderBean(leader, zks);
+    }
+
+    @After
+    public void tearDown() throws IOException {
+        fileTxnSnapLog.close();
+    }
+
+    @Test
+    public void testGetName() {
+        assertEquals("Leader", leaderBean.getName());
+    }
+
+    @Test
+    public void testGetCurrentZxid() {
+        // Arrange
+        zks.setZxid(1);
+
+        // Assert
+        assertEquals("0x1", leaderBean.getCurrentZxid());
+    }
+
+    @Test
+    public void testGetElectionTimeTaken() {
+        // Arrange
+        qp.setElectionTimeTaken(1);
+
+        // Assert
+        assertEquals(1, leaderBean.getElectionTimeTaken());
+    }
+
+    @Test
+    public void testGetProposalSize() throws IOException, Leader.XidRolloverException {
+        // Arrange
+        Request req = createMockRequest();
+
+        // Act
+        leader.propose(req);
+
+        // Assert
+        byte[] data = SerializeUtils.serializeRequest(req);
+        assertEquals(data.length, leaderBean.getLastProposalSize());
+        assertEquals(data.length, leaderBean.getMinProposalSize());
+        assertEquals(data.length, leaderBean.getMaxProposalSize());
+    }
+
+    @Test
+    public void testResetProposalStats() throws IOException, Leader.XidRolloverException {
+        // Arrange
+        int initialProposalSize = leaderBean.getLastProposalSize();
+        Request req = createMockRequest();
+
+        // Act
+        leader.propose(req);
+
+        // Assert
+        assertNotEquals(initialProposalSize, leaderBean.getLastProposalSize());
+        leaderBean.resetProposalStatistics();
+        assertEquals(initialProposalSize, leaderBean.getLastProposalSize());
+        assertEquals(initialProposalSize, leaderBean.getMinProposalSize());
+        assertEquals(initialProposalSize, leaderBean.getMaxProposalSize());
+    }
+
+    private Request createMockRequest() throws IOException {
+        TxnHeader header = mock(TxnHeader.class);
+        doAnswer(new Answer() {
+            @Override
+            public Object answer(InvocationOnMock invocation) throws Throwable {
+                Object[] args = invocation.getArguments();
+                OutputArchive oa = (OutputArchive) args[0];
+                oa.writeString("header", "test");
+                return null;
+            }
+        }).when(header).serialize(any(OutputArchive.class), anyString());
+        Record txn = mock(Record.class);
+        doAnswer(new Answer() {
+            @Override
+            public Object answer(InvocationOnMock invocation) throws Throwable {
+                Object[] args = invocation.getArguments();
+                OutputArchive oa = (OutputArchive) args[0];
+                oa.writeString("record", "test");
+                return null;
+            }
+        }).when(txn).serialize(any(OutputArchive.class), anyString());
+        return new Request(1, 2, 3, header, txn, 4);
+    }
+}

src/java/test/org/apache/zookeeper/server/quorum/ProposalStatsTest.java

@@ -0,0 +1,58 @@
+/**
+ * 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.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class ProposalStatsTest {
+    @Test
+    public void testSetProposalSizeSetMinMax() {
+        ProposalStats stats = new ProposalStats();
+        assertEquals(-1, stats.getLastProposalSize());
+        assertEquals(-1, stats.getMinProposalSize());
+        assertEquals(-1, stats.getMaxProposalSize());
+        stats.setLastProposalSize(10);
+        assertEquals(10, stats.getLastProposalSize());
+        assertEquals(10, stats.getMinProposalSize());
+        assertEquals(10, stats.getMaxProposalSize());
+        stats.setLastProposalSize(20);
+        assertEquals(20, stats.getLastProposalSize());
+        assertEquals(10, stats.getMinProposalSize());
+        assertEquals(20, stats.getMaxProposalSize());
+        stats.setLastProposalSize(5);
+        assertEquals(5, stats.getLastProposalSize());
+        assertEquals(5, stats.getMinProposalSize());
+        assertEquals(20, stats.getMaxProposalSize());
+    }
+
+    @Test
+    public void testReset() {
+        ProposalStats stats = new ProposalStats();
+        stats.setLastProposalSize(10);
+        assertEquals(10, stats.getLastProposalSize());
+        assertEquals(10, stats.getMinProposalSize());
+        assertEquals(10, stats.getMaxProposalSize());
+        stats.reset();
+        assertEquals(-1, stats.getLastProposalSize());
+        assertEquals(-1, stats.getMinProposalSize());
+        assertEquals(-1, stats.getMaxProposalSize());
+    }
+}

src/java/test/org/apache/zookeeper/server/quorum/StatCommandTest.java

@@ -0,0 +1,106 @@
+/**
+ * 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.server.ServerCnxn;
+import org.apache.zookeeper.server.ServerCnxnFactory;
+import org.apache.zookeeper.server.ServerStats;
+import org.apache.zookeeper.server.ZKDatabase;
+import org.apache.zookeeper.server.command.FourLetterCommands;
+import org.apache.zookeeper.server.command.StatCommand;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class StatCommandTest {
+    private StringWriter outputWriter;
+    private StatCommand statCommand;
+    private ServerStats.Provider providerMock;
+
+    @Before
+    public void setUp() throws IOException {
+        outputWriter = new StringWriter();
+        ServerCnxn serverCnxnMock = mock(ServerCnxn.class);
+
+        LeaderZooKeeperServer zks = mock(LeaderZooKeeperServer.class);
+        when(zks.isRunning()).thenReturn(true);
+        providerMock = mock(ServerStats.Provider.class);
+        when(zks.serverStats()).thenReturn(new ServerStats(providerMock));
+        ZKDatabase zkDatabaseMock = mock(ZKDatabase.class);
+        when(zks.getZKDatabase()).thenReturn(zkDatabaseMock);
+        Leader leaderMock = mock(Leader.class);
+        when(leaderMock.getProposalStats()).thenReturn(new ProposalStats());
+        when(zks.getLeader()).thenReturn(leaderMock);
+
+        ServerCnxnFactory serverCnxnFactory = mock(ServerCnxnFactory.class);
+        ServerCnxn serverCnxn = mock(ServerCnxn.class);
+        List<ServerCnxn> connections = new ArrayList<>();
+        connections.add(serverCnxn);
+        when(serverCnxnFactory.getConnections()).thenReturn(connections);
+
+        statCommand = new StatCommand(new PrintWriter(outputWriter), serverCnxnMock, FourLetterCommands.statCmd);
+        statCommand.setZkServer(zks);
+        statCommand.setFactory(serverCnxnFactory);
+    }
+
+    @Test
+    public void testLeaderStatCommand() {
+        // Arrange
+        when(providerMock.getState()).thenReturn("leader");
+
+        // Act
+        statCommand.commandRun();
+
+        // Assert
+        String output = outputWriter.toString();
+        assertCommonStrings(output);
+        assertThat(output, containsString("Mode: leader"));
+        assertThat(output, containsString("Proposal sizes last/min/max:"));
+    }
+
+    @Test
+    public void testFollowerStatCommand() {
+        // Arrange
+        when(providerMock.getState()).thenReturn("follower");
+
+        // Act
+        statCommand.commandRun();
+
+        // Assert
+        String output = outputWriter.toString();
+        assertCommonStrings(output);
+        assertThat(output, containsString("Mode: follower"));
+    }
+
+    private void assertCommonStrings(String output) {
+        assertThat(output, containsString("Clients:"));
+        assertThat(output, containsString("Zookeeper version:"));
+        assertThat(output, containsString("Node count:"));
+    }
+}

src/java/test/org/apache/zookeeper/server/quorum/StatResetCommandTest.java

@@ -0,0 +1,111 @@
+/**
+ * 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.server.ServerCnxn;
+import org.apache.zookeeper.server.ServerStats;
+import org.apache.zookeeper.server.ZooKeeperServer;
+import org.apache.zookeeper.server.command.StatResetCommand;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+
+import static org.apache.zookeeper.server.command.AbstractFourLetterCommand.ZK_NOT_SERVING;
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class StatResetCommandTest {
+    private StatResetCommand statResetCommand;
+    private StringWriter outputWriter;
+    private ZooKeeperServer zks;
+    private ServerStats serverStats;
+
+    @Before
+    public void setUp() {
+        outputWriter = new StringWriter();
+        ServerCnxn serverCnxnMock = mock(ServerCnxn.class);
+
+        zks = mock(ZooKeeperServer.class);
+        when(zks.isRunning()).thenReturn(true);
+
+        serverStats = mock(ServerStats.class);
+        when(zks.serverStats()).thenReturn(serverStats);
+
+        statResetCommand = new StatResetCommand(new PrintWriter(outputWriter), serverCnxnMock);
+        statResetCommand.setZkServer(zks);
+    }
+
+    @Test
+    public void testStatResetWithZKNotRunning() {
+        // Arrange
+        when(zks.isRunning()).thenReturn(false);
+
+        // Act
+        statResetCommand.commandRun();
+
+        // Assert
+        String output = outputWriter.toString();
+        assertEquals(ZK_NOT_SERVING + "\n", output);
+    }
+
+    @Test
+    public void testStatResetWithFollower() {
+        // Arrange
+        when(zks.isRunning()).thenReturn(true);
+        when(serverStats.getServerState()).thenReturn("follower");
+
+        // Act
+        statResetCommand.commandRun();
+
+        // Assert
+        String output = outputWriter.toString();
+        assertEquals("Server stats reset.\n", output);
+        verify(serverStats, times(1)).reset();
+    }
+
+    @Test
+    public void testStatResetWithLeader() {
+        // Arrange
+        LeaderZooKeeperServer leaderZks = mock(LeaderZooKeeperServer.class);
+        when(leaderZks.isRunning()).thenReturn(true);
+        when(leaderZks.serverStats()).thenReturn(serverStats);
+        Leader leader = mock(Leader.class);
+        when(leaderZks.getLeader()).thenReturn(leader);
+        statResetCommand.setZkServer(leaderZks);
+
+        when(serverStats.getServerState()).thenReturn("leader");
+
+        ProposalStats proposalStats = mock(ProposalStats.class);
+        when(leader.getProposalStats()).thenReturn(proposalStats);
+
+        // Act
+        statResetCommand.commandRun();
+
+        // Assert
+        String output = outputWriter.toString();
+        assertEquals("Server stats reset.\n", output);
+        verify(serverStats, times(1)).reset();
+        verify(proposalStats, times(1)).reset();
+    }
+}

src/java/test/org/apache/zookeeper/server/util/SerializeUtilsTest.java

@@ -0,0 +1,128 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.util;
+
+import org.apache.jute.BinaryOutputArchive;
+import org.apache.jute.OutputArchive;
+import org.apache.jute.Record;
+import org.apache.zookeeper.server.Request;
+import org.apache.zookeeper.txn.TxnHeader;
+import org.junit.Test;
+import org.mockito.InOrder;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+public class SerializeUtilsTest {
+
+    @Test
+    public void testSerializeRequestRequestIsNull() {
+        byte[] data = SerializeUtils.serializeRequest(null);
+        assertNull(data);
+    }
+
+    @Test
+    public void testSerializeRequestRequestHeaderIsNull() {
+        Request request = new Request(0, 0, 0, null, null, 0);
+        byte[] data = SerializeUtils.serializeRequest(request);
+        assertNull(data);
+    }
+
+    @Test
+    public void testSerializeRequestWithoutTxn() throws IOException {
+        // Arrange
+        TxnHeader header = mock(TxnHeader.class);
+        doAnswer(new Answer() {
+            @Override
+            public Object answer(InvocationOnMock invocation) throws Throwable {
+                Object[] args = invocation.getArguments();
+                OutputArchive oa = (OutputArchive) args[0];
+                oa.writeString("header", "test");
+                return null;
+            }
+        }).when(header).serialize(any(OutputArchive.class), anyString());
+        Request request = new Request(1, 2, 3, header, null, 4);
+
+        // Act
+        byte[] data = SerializeUtils.serializeRequest(request);
+
+        // Assert
+        assertNotNull(data);
+        verify(header).serialize(any(OutputArchive.class), eq("hdr"));
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
+        boa.writeString("header", "test");
+        baos.close();
+        assertArrayEquals(baos.toByteArray(), data);
+    }
+
+    @Test
+    public void testSerializeRequestWithTxn() throws IOException {
+        // Arrange
+        TxnHeader header = mock(TxnHeader.class);
+        doAnswer(new Answer() {
+            @Override
+            public Object answer(InvocationOnMock invocation) throws Throwable {
+                Object[] args = invocation.getArguments();
+                OutputArchive oa = (OutputArchive) args[0];
+                oa.writeString("header", "test");
+                return null;
+            }
+        }).when(header).serialize(any(OutputArchive.class), anyString());
+        Record txn = mock(Record.class);
+        doAnswer(new Answer() {
+            @Override
+            public Object answer(InvocationOnMock invocation) throws Throwable {
+                Object[] args = invocation.getArguments();
+                OutputArchive oa = (OutputArchive) args[0];
+                oa.writeString("record", "test");
+                return null;
+            }
+        }).when(txn).serialize(any(OutputArchive.class), anyString());
+        Request request = new Request(1, 2, 3, header, txn, 4);
+
+        // Act
+        byte[] data = SerializeUtils.serializeRequest(request);
+
+        // Assert
+        assertNotNull(data);
+        InOrder inOrder = inOrder(header, txn);
+        inOrder.verify(header).serialize(any(OutputArchive.class), eq("hdr"));
+        inOrder.verify(txn).serialize(any(OutputArchive.class), eq("txn"));
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
+        boa.writeString("header", "test");
+        boa.writeString("record", "test");
+        baos.close();
+        assertArrayEquals(baos.toByteArray(), data);
+    }
+}