ZOOKEEPER-3066: Expose on JMX of Followers the id of the current leader

Expose a new JMX attribute "isLeader" in RemotePeerBean and LocalPeerBean

Author: Enrico Olivelli <eolivelli@apache.org>

Reviewers: Andor Molnar <andor@apache.org>

Closes #546 from eolivelli/fix/ZOOKEEPER-3066-leader-jmx and squashes the following commits:

bf09e37d [Enrico Olivelli] clean up spaces
ded4ce20 [Enrico Olivelli] reorder imports
81d7bb1c [Enrico Olivelli] split tests into LocalPeerBeanTest and QuorumPeerTest
a1984ef1 [Enrico Olivelli] fix typo
eb8d1bea [Enrico Olivelli] split tests
86b116b0 [Enrico Olivelli] enhance tests
b5000efc [Enrico Olivelli] clean up test
43782fe5 [Enrico Olivelli] introduce mockito tests
2778e742 [Enrico Olivelli] revert "ZOOKEEPER-3066 add servers restart to test case"
5f120631 [Enrico Olivelli] ZOOKEEPER-3066 add servers restart to test case
067aed9e [Enrico Olivelli] ZOOKEEPER-3066 Expose on JMX of Followers the id of the current leader Enhance RemotePeerMXBean and LocalPeerMXBean by adding information about leadership

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

@@ -109,4 +109,9 @@ public class LocalPeerBean extends ServerBean implements LocalPeerMXBean {
     public boolean isPartOfEnsemble() {
         return peer.getView().containsKey(peer.getId());
     }
+
+    @Override
+    public boolean isLeader() {
+        return peer.isLeader(peer.getId());
+    }
 }

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

@@ -104,4 +104,9 @@ public interface LocalPeerMXBean extends ServerMXBean {
      * @return true if quorum peer is part of the ensemble, false otherwise
      */
     public boolean isPartOfEnsemble();
+
+    /**
+     * @return true if the peer is the current leader
+     */
+    public boolean isLeader();
 }

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

@@ -148,6 +148,10 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
         public QuorumServer(long id, InetSocketAddress addr) {
             this(id, addr, (InetSocketAddress)null, (InetSocketAddress)null, LearnerType.PARTICIPANT);
         }
+ 
+        public long getId() {
+            return id;
+        }
 
         /**
          * Performs a DNS lookup for server address and election address.
@@ -466,6 +470,11 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
         return myid;
     }
 
+    // VisibleForTesting
+    void setId(long id) {
+        this.myid = id;
+    }
+
     /**
      * This is who I think the leader currently is.
      */
@@ -1093,7 +1102,7 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
                         jmxLocalPeerBean = null;
                     }
                 } else {
-                    RemotePeerBean rBean = new RemotePeerBean(s);
+                    RemotePeerBean rBean = new RemotePeerBean(this, s);
                     try {
                         MBeanRegistry.getInstance().register(rBean, jmxQuorumBean);
                         jmxRemotePeerBean.put(s.id, rBean);
@@ -1895,7 +1904,7 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
         joiningMembers.remove(getId()); // remove self as it is local bean
         for (Long id : joiningMembers) {
             QuorumServer qs = newMembers.get(id);
-            RemotePeerBean rBean = new RemotePeerBean(qs);
+            RemotePeerBean rBean = new RemotePeerBean(this, qs);
             try {
                 MBeanRegistry.getInstance().register(rBean, jmxQuorumBean);
                 jmxRemotePeerBean.put(qs.id, rBean);
@@ -2069,4 +2078,9 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
                 this.quorumCnxnThreadsSize,
                 this.isQuorumSaslAuthEnabled());
     }
+
+    boolean isLeader(long id) {
+        Vote vote = getCurrentVote();
+        return vote != null && id == vote.getId();
+    }
 }

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

@@ -26,9 +26,11 @@ import org.apache.zookeeper.jmx.ZKMBeanInfo;
  */
 public class RemotePeerBean implements RemotePeerMXBean,ZKMBeanInfo {
     private QuorumPeer.QuorumServer peer;
-    
-    public RemotePeerBean(QuorumPeer.QuorumServer peer){
+    private final QuorumPeer localPeer;
+
+    public RemotePeerBean(QuorumPeer localPeer, QuorumPeer.QuorumServer peer){
         this.peer=peer;
+        this.localPeer = localPeer;
     }
 
     public void setQuorumServer(QuorumPeer.QuorumServer peer) {
@@ -61,4 +63,10 @@ public class RemotePeerBean implements RemotePeerMXBean,ZKMBeanInfo {
     public String getLearnerType() {
         return peer.type.toString();
     }
+
+    @Override
+    public boolean isLeader() {
+        return localPeer.isLeader(peer.getId());
+    }
+    
 }

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

@@ -45,4 +45,9 @@ public interface RemotePeerMXBean {
      * @return the learner type
      */
     public String getLearnerType();
+
+    /**
+     * @return true if the peer is the current leader
+     */
+    public boolean isLeader();
 }

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

@@ -20,11 +20,14 @@ package org.apache.zookeeper.server.quorum;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
-
 import org.apache.zookeeper.PortAssignment;
 import org.apache.zookeeper.server.ServerCnxnFactory;
 import org.junit.Test;
@@ -79,4 +82,24 @@ public class LocalPeerBeanTest {
         cnxnFactory.shutdown();
     }
 
+    @Test
+    public void testLocalPeerIsLeader() throws Exception {
+        long localPeerId = 7;
+        QuorumPeer peer = mock(QuorumPeer.class);
+        when(peer.getId()).thenReturn(localPeerId);
+        when(peer.isLeader(eq(localPeerId))).thenReturn(true);
+        LocalPeerBean localPeerBean = new LocalPeerBean(peer);
+        assertTrue(localPeerBean.isLeader());
+    }
+
+    @Test
+    public void testLocalPeerIsNotLeader() throws Exception {
+        long localPeerId = 7;
+        QuorumPeer peer = mock(QuorumPeer.class);
+        when(peer.getId()).thenReturn(localPeerId);
+        when(peer.isLeader(eq(localPeerId))).thenReturn(false);
+        LocalPeerBean localPeerBean = new LocalPeerBean(peer);
+        assertFalse(localPeerBean.isLeader());
+    }
+
 }

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

@@ -84,4 +84,34 @@ public class QuorumPeerTest {
         peer2.shutdown();
     }
 
+    @Test
+    public void testLocalPeerIsLeader() throws Exception {
+        long localPeerId = 7;
+        QuorumPeer peer = new QuorumPeer();
+        peer.setId(localPeerId);
+        Vote voteLocalPeerIsLeader = new Vote(localPeerId, 0);
+        peer.setCurrentVote(voteLocalPeerIsLeader);
+        assertTrue(peer.isLeader(localPeerId));
+    }
+
+    @Test
+    public void testLocalPeerIsNotLeader() throws Exception {
+        long localPeerId = 7;
+        long otherPeerId = 17;
+        QuorumPeer peer = new QuorumPeer();
+        peer.setId(localPeerId);
+        Vote voteLocalPeerIsNotLeader = new Vote(otherPeerId, 0);
+        peer.setCurrentVote(voteLocalPeerIsNotLeader);
+        assertFalse(peer.isLeader(localPeerId));
+    }
+
+    @Test
+    public void testIsNotLeaderBecauseNoVote() throws Exception {
+        long localPeerId = 7;
+        QuorumPeer peer = new QuorumPeer();
+        peer.setId(localPeerId);
+        peer.setCurrentVote(null);
+        assertFalse(peer.isLeader(localPeerId));
+    }
+
 }

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

@@ -19,7 +19,12 @@
 package org.apache.zookeeper.server.quorum;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.mockito.Matchers.eq;
 
 import java.net.InetSocketAddress;
 
@@ -36,10 +41,24 @@ public class RemotePeerBeanTest {
         InetSocketAddress peerCommunicationAddress = null;
         // Here peerCommunicationAddress is null, also clientAddr is null
         QuorumServer peer = new QuorumServer(1, peerCommunicationAddress);
-        RemotePeerBean remotePeerBean = new RemotePeerBean(peer);
+        RemotePeerBean remotePeerBean = new RemotePeerBean(null, peer);
         String clientAddress = remotePeerBean.getClientAddress();
         assertNotNull(clientAddress);
         assertEquals(0, clientAddress.length());
     }
 
+    @Test
+    @SuppressWarnings("unchecked")
+    public void testIsLeader() {
+        long peerId = 7;
+        QuorumPeer.QuorumServer quorumServerMock = mock(QuorumPeer.QuorumServer.class);
+        when(quorumServerMock.getId()).thenReturn(peerId);
+        QuorumPeer peerMock = mock(QuorumPeer.class);
+        RemotePeerBean remotePeerBean = new RemotePeerBean(peerMock, quorumServerMock);
+        when(peerMock.isLeader(eq(peerId))).thenReturn(true);
+        assertTrue(remotePeerBean.isLeader());
+        when(peerMock.isLeader(eq(peerId))).thenReturn(false);
+        assertFalse(remotePeerBean.isLeader());
+    }
+
 }

src/java/test/org/apache/zookeeper/test/HierarchicalQuorumTest.java

@@ -226,30 +226,30 @@ public class HierarchicalQuorumTest extends ClientBase {
                                     CONNECTION_TIMEOUT));
             LOG.info(hp + " is accepting client connections");
         }
-
+        final int numberOfPeers = 5;
         // interesting to see what's there...
         JMXEnv.dump();
         // make sure we have these 5 servers listed
         Set<String> ensureNames = new LinkedHashSet<String>();
-        for (int i = 1; i <= 5; i++) {
+        for (int i = 1; i <= numberOfPeers; i++) {
             ensureNames.add("InMemoryDataTree");
         }
-        for (int i = 1; i <= 5; i++) {
+        for (int i = 1; i <= numberOfPeers; i++) {
             ensureNames.add("name0=ReplicatedServer_id" + i
                  + ",name1=replica." + i + ",name2=");
         }
-        for (int i = 1; i <= 5; i++) {
-            for (int j = 1; j <= 5; j++) {
+        for (int i = 1; i <= numberOfPeers; i++) {
+            for (int j = 1; j <= numberOfPeers; j++) {
                 ensureNames.add("name0=ReplicatedServer_id" + i
                      + ",name1=replica." + j);
             }
         }
-        for (int i = 1; i <= 5; i++) {
+        for (int i = 1; i <= numberOfPeers; i++) {
             ensureNames.add("name0=ReplicatedServer_id" + i);
         }
         JMXEnv.ensureAll(ensureNames.toArray(new String[ensureNames.size()]));
-
-        for (int i = 1; i <= 5; i++) {
+        for (int i = 1; i <= numberOfPeers; i++) {
+            // LocalPeerBean
             String bean = CommonNames.DOMAIN + ":name0=ReplicatedServer_id" + i
                     + ",name1=replica." + i;
             JMXEnv.ensureBeanAttribute(bean, "ConfigVersion");
@@ -257,6 +257,23 @@ public class HierarchicalQuorumTest extends ClientBase {
             JMXEnv.ensureBeanAttribute(bean, "ClientAddress");
             JMXEnv.ensureBeanAttribute(bean, "ElectionAddress");
             JMXEnv.ensureBeanAttribute(bean, "QuorumSystemInfo");
+            JMXEnv.ensureBeanAttribute(bean, "Leader");
+        }
+
+        for (int i = 1; i <= numberOfPeers; i++) {
+            for (int j = 1; j <= numberOfPeers; j++) {
+                if (j != i) {
+                    // RemotePeerBean
+                    String bean = CommonNames.DOMAIN + ":name0=ReplicatedServer_id" + i
+                            + ",name1=replica." + j;
+                    JMXEnv.ensureBeanAttribute(bean, "Name");
+                    JMXEnv.ensureBeanAttribute(bean, "LearnerType");
+                    JMXEnv.ensureBeanAttribute(bean, "ClientAddress");
+                    JMXEnv.ensureBeanAttribute(bean, "ElectionAddress");
+                    JMXEnv.ensureBeanAttribute(bean, "QuorumAddress");
+                    JMXEnv.ensureBeanAttribute(bean, "Leader");
+                }
+            }
         }
     }