Use the electionId reported by ismaster to detect a stale primary by invalidate primaries that have an older electionId than one already seen.

jyemin · jyemin · commit f399388c63f9 · 2015-05-10T21:03:20.000-04:00
JAVA-1799
diff --git a/src/main/com/mongodb/MultiServerCluster.java b/src/main/com/mongodb/MultiServerCluster.java
@@ -16,6 +16,8 @@
 
 package com.mongodb;
 
+import org.bson.types.ObjectId;
+
 import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.HashSet;
@@ -42,6 +44,7 @@ final class MultiServerCluster extends BaseCluster {
 
     private ClusterType clusterType;
     private String replicaSetName;
+    private ObjectId maxElectionId;
     private final ConcurrentMap<ServerAddress, ServerTuple> addressToServerTupleMap =
     new ConcurrentHashMap<ServerAddress, ServerTuple>();
 
@@ -117,6 +120,7 @@ private void onChange(final ChangeEvent<ServerDescription> event) {
             return;
         }
 
+        boolean shouldUpdateDescription = true;
         synchronized (this) {
             ServerDescription newDescription = event.getNewValue();
             ServerTuple serverTuple = addressToServerTupleMap.get(newDescription.getAddress());
@@ -132,36 +136,40 @@ private void onChange(final ChangeEvent<ServerDescription> event) {
 
                 switch (clusterType) {
                     case ReplicaSet:
-                        handleReplicaSetMemberChanged(newDescription);
+                        shouldUpdateDescription = handleReplicaSetMemberChanged(newDescription);
                         break;
                     case Sharded:
-                        handleShardRouterChanged(newDescription);
+                        shouldUpdateDescription = handleShardRouterChanged(newDescription);
                         break;
                     case StandAlone:
-                        handleStandAloneChanged(newDescription);
+                        shouldUpdateDescription = handleStandAloneChanged(newDescription);
                         break;
                     default:
                         break;
                 }
             }
 
-            serverTuple.description = newDescription;
-            updateDescription();
+            if (shouldUpdateDescription) {
+                serverTuple.description = newDescription;
+                updateDescription();
+            }
+        }
+        if (shouldUpdateDescription) {
+            fireChangeEvent();
         }
-        fireChangeEvent();
     }
 
-    private void handleReplicaSetMemberChanged(final ServerDescription newDescription) {
+    private boolean handleReplicaSetMemberChanged(final ServerDescription newDescription) {
         if (!newDescription.isReplicaSetMember()) {
             LOGGER.severe(format("Expecting replica set member, but found a %s.  Removing %s from client view of cluster.",
                                  newDescription.getType(), newDescription.getAddress()));
             removeServer(newDescription.getAddress());
-            return;
+            return true;
         }
 
         if (newDescription.getType() == ReplicaSetGhost) {
             LOGGER.info(format("Server %s does not appear to be a member of an initiated replica set.", newDescription.getAddress()));
-            return;
+            return true;
         }
 
         if (replicaSetName == null) {
@@ -174,39 +182,51 @@ private void handleReplicaSetMemberChanged(final ServerDescription newDescriptio
                                  replicaSetName, newDescription.getSetName(), newDescription.getAddress()
                                 ));
             removeServer(newDescription.getAddress());
-            return;
+            return true;
         }
 
         ensureServers(newDescription);
 
         if (newDescription.isPrimary()) {
+            if (newDescription.getElectionId() != null) {
+                if (maxElectionId != null && maxElectionId.compareTo(newDescription.getElectionId()) > 0) {
+                    addressToServerTupleMap.get(newDescription.getAddress()).server.invalidate();
+                    return false;
+                }
+
+                maxElectionId = newDescription.getElectionId();
+            }
+
             if (isNotAlreadyPrimary(newDescription.getAddress())) {
                 LOGGER.info(format("Discovered replica set primary %s", newDescription.getAddress()));
             }
             invalidateOldPrimaries(newDescription.getAddress());
         }
+        return true;
     }
 
     private boolean isNotAlreadyPrimary(final ServerAddress address) {
         ServerTuple serverTuple = addressToServerTupleMap.get(address);
         return serverTuple == null || !serverTuple.description.isPrimary();
     }
 
-    private void handleShardRouterChanged(final ServerDescription newDescription) {
+    private boolean handleShardRouterChanged(final ServerDescription newDescription) {
         if (newDescription.getClusterType() != Sharded) {
             LOGGER.severe(format("Expecting a %s, but found a %s.  Removing %s from client view of cluster.",
                                  ShardRouter, newDescription.getType(), newDescription.getAddress()));
             removeServer(newDescription.getAddress());
         }
+        return true;
     }
 
-    private void handleStandAloneChanged(final ServerDescription newDescription) {
+    private boolean handleStandAloneChanged(final ServerDescription newDescription) {
         if (getSettings().getHosts().size() > 1) {
             LOGGER.severe(format("Expecting a single %s, but found more than one.  Removing %s from client view of cluster.",
                                  ServerType.StandAlone, newDescription.getAddress()));
             clusterType = Unknown;
             removeServer(newDescription.getAddress());
         }
+        return true;
     }
 
     private void addServer(final ServerAddress serverAddress) {
diff --git a/src/main/com/mongodb/ServerDescription.java b/src/main/com/mongodb/ServerDescription.java
@@ -16,6 +16,7 @@
 
 package com.mongodb;
 
+import org.bson.types.ObjectId;
 import org.bson.util.annotations.Immutable;
 
 import java.text.DecimalFormat;
@@ -71,6 +72,8 @@ class ServerDescription {
     private final int minWireVersion;
     private final int maxWireVersion;
 
+    private final ObjectId electionId;
+
     private Throwable exception;
 
     static class Builder {
@@ -91,6 +94,7 @@ static class Builder {
         private ServerVersion version = new ServerVersion();
         private int minWireVersion = 0;
         private int maxWireVersion = 0;
+        private ObjectId electionId;
         private Throwable exception;
 
         // CHECKSTYLE:OFF
@@ -180,6 +184,17 @@ public Builder maxWireVersion(final int maxWireVersion) {
             return this;
         }
 
+        /**
+         * Sets the electionId reported by this server.
+         *
+         * @param electionId the electionId
+         * @return this
+         */
+        public Builder electionId(final ObjectId electionId) {
+            this.electionId = electionId;
+            return this;
+        }
+
         public Builder exception(final Throwable exception) {
             this.exception = exception;
             return this;
@@ -302,6 +317,15 @@ public int getMaxWireVersion() {
         return maxWireVersion;
     }
 
+    /**
+     * The replica set electionid reported by this MongoDB server.
+     *
+     * @return the electionId, which may be null
+     */
+    public ObjectId getElectionId() {
+        return electionId;
+    }
+
     /**
      * Returns true if the server has the given tags.  A server of either type {@code ServerType.StandAlone} or
      * {@code ServerType.ShardRouter} is considered to have all tags, so this method will always return true for instances of either of
@@ -420,6 +444,9 @@ public boolean equals(final Object o) {
         if (maxWireVersion != that.maxWireVersion) {
             return false;
         }
+        if (electionId != null ? !electionId.equals(that.electionId) : that.electionId != null) {
+            return false;
+        }
 
         // Compare class equality and message as exceptions rarely override equals
         Class thisExceptionClass = exception != null ? exception.getClass() : null;
@@ -456,6 +483,7 @@ public int hashCode() {
         result = 31 * result + version.hashCode();
         result = 31 * result + minWireVersion;
         result = 31 * result + maxWireVersion;
+        result = 31 * result + (electionId != null ? electionId.hashCode() : 0);
         result = 31 * result + (exception == null ? 0 : exception.getClass().hashCode());
         result = 31 * result + (exception == null ? 0 : exception.getMessage().hashCode());
         return result;
@@ -473,6 +501,7 @@ public String toString() {
                + ", maxDocumentSize=" + maxDocumentSize
                + ", maxMessageSize=" + maxMessageSize
                + ", maxWriteBatchSize=" + maxWriteBatchSize
+               + ", electionId=" + electionId
                + ", tagSet=" + tagSet
                + ", setName='" + setName + '\''
                + ", averageLatencyNanos=" + averageLatencyNanos
@@ -534,6 +563,7 @@ private String getAverageLatencyFormattedInMilliseconds() {
         ok = builder.ok;
         minWireVersion = builder.minWireVersion;
         maxWireVersion = builder.maxWireVersion;
+        electionId = builder.electionId;
         exception = builder.exception;
     }
 }
diff --git a/src/main/com/mongodb/ServerMonitor.java b/src/main/com/mongodb/ServerMonitor.java
@@ -257,6 +257,7 @@ private ServerDescription createDescription(final CommandResult commandResult, f
                                 .setName(commandResult.getString("setName"))
                                 .minWireVersion(commandResult.getInt("minWireVersion", ServerDescription.getDefaultMinWireVersion()))
                                 .maxWireVersion(commandResult.getInt("maxWireVersion", ServerDescription.getDefaultMaxWireVersion()))
+                                .electionId(commandResult.containsKey("electionId") ? commandResult.getObjectId("electionId") : null)
                                 .averageLatency(averageLatencyNanos, TimeUnit.NANOSECONDS)
                                 .ok(commandResult.ok()).build();
     }
@@ -319,4 +320,4 @@ private ServerDescription getConnectingServerDescription(final Throwable throwab
                                 .exception(throwable)
                                 .build();
     }
-}
+}
diff --git a/src/test/com/mongodb/MultiServerClusterSpecification.groovy b/src/test/com/mongodb/MultiServerClusterSpecification.groovy
@@ -16,6 +16,7 @@
 
 package com.mongodb
 
+import org.bson.types.ObjectId
 import spock.lang.Specification
 
 import static com.mongodb.ClusterConnectionMode.Multiple
@@ -259,6 +260,22 @@ class MultiServerClusterSpecification extends Specification {
         getClusterDescription(cluster).all == getServerDescriptions(firstServer, secondServer, thirdServer)
     }
 
+    def 'should invalidate new primary if its electionId is less than the previously reported electionId'() {
+        given:
+        def cluster = new MultiServerCluster(CLUSTER_ID, ClusterSettings.builder().hosts([firstServer, secondServer]).build(), factory,
+                                             CLUSTER_LISTENER)
+        sendNotification(firstServer, ReplicaSetPrimary, [firstServer, secondServer, thirdServer], new ObjectId(new Date(1000)))
+
+        when:
+        sendNotification(secondServer, ReplicaSetPrimary, [firstServer, secondServer, thirdServer], new ObjectId(new Date(999)))
+
+        then:
+        getServerDescription(firstServer).state == Connected
+        getServerDescription(firstServer).type == ReplicaSetPrimary
+        getServerDescription(secondServer).state == Connecting
+        getClusterDescription(cluster).all == getServerDescriptions(firstServer, secondServer, thirdServer)
+    }
+
     def 'should remove a server when a server in the seed list is not in hosts list, it should be removed'() {
         given:
         def serverAddressAlias = new ServerAddress('alternate')
@@ -465,11 +482,17 @@ class MultiServerClusterSpecification extends Specification {
 
     def sendNotification(ServerAddress serverAddress, ServerType serverType, List<ServerAddress> hosts, List<ServerAddress> passives,
                          String setName) {
-        factory.getServer(serverAddress).sendNotification(getBuilder(serverAddress, serverType, hosts, passives, true, setName).build())
+        factory.getServer(serverAddress).sendNotification(getBuilder(serverAddress, serverType, hosts, passives, true, setName, null)
+                                                                  .build())
+    }
+
+    def sendNotification(ServerAddress serverAddress, ServerType serverType, List<ServerAddress> hosts, ObjectId electionId) {
+        factory.getServer(serverAddress).sendNotification(getBuilder(serverAddress, serverType, hosts, [], true, 'test', electionId)
+                                                                  .build())
     }
 
     def sendNotification(ServerAddress serverAddress, ServerType serverType, List<ServerAddress> hosts, boolean ok) {
-        factory.getServer(serverAddress).sendNotification(getBuilder(serverAddress, serverType, hosts, [], ok, null).build())
+        factory.getServer(serverAddress).sendNotification(getBuilder(serverAddress, serverType, hosts, [], ok, null, null).build())
     }
 
     def getClusterDescription(MultiServerCluster cluster) {
@@ -485,7 +508,7 @@ class MultiServerClusterSpecification extends Specification {
     }
 
     def getBuilder(ServerAddress serverAddress, ServerType serverType, List<ServerAddress> hosts, List<ServerAddress> passives, boolean ok,
-                   String setName) {
+                   String setName, ObjectId electionId) {
         ServerDescription.builder()
                          .address(serverAddress)
                          .type(serverType)
@@ -494,5 +517,6 @@ class MultiServerClusterSpecification extends Specification {
                          .hosts(hosts*.toString() as Set)
                          .passives(passives*.toString() as Set)
                          .setName(setName)
+                         .electionId(electionId)
     }
 }
diff --git a/src/test/com/mongodb/ServerDescriptionTest.java b/src/test/com/mongodb/ServerDescriptionTest.java
@@ -17,6 +17,7 @@
 package com.mongodb;
 
 
+import org.bson.types.ObjectId;
 import org.junit.Test;
 
 import java.net.UnknownHostException;
@@ -85,6 +86,7 @@ public void testDefaults() throws UnknownHostException {
         assertEquals(new ServerVersion(), serverDescription.getVersion());
         assertEquals(0, serverDescription.getMinWireVersion());
         assertEquals(0, serverDescription.getMaxWireVersion());
+        assertNull(serverDescription.getElectionId());
         assertNull(serverDescription.getException());
     }
 
@@ -112,6 +114,7 @@ public void testBuilder() throws UnknownHostException {
                                               .version(new ServerVersion(asList(2, 4, 1)))
                                               .minWireVersion(1)
                                               .maxWireVersion(2)
+                                              .electionId(new ObjectId("123412341234123412341234"))
                                               .exception(exception)
                                               .build();
 
@@ -144,6 +147,7 @@ public void testBuilder() throws UnknownHostException {
         assertEquals(new ServerVersion(asList(2, 4, 1)), serverDescription.getVersion());
         assertEquals(1, serverDescription.getMinWireVersion());
         assertEquals(2, serverDescription.getMaxWireVersion());
+        assertEquals(new ObjectId("123412341234123412341234"), serverDescription.getElectionId());
         assertEquals(exception, serverDescription.getException());
     }
 
@@ -167,6 +171,7 @@ public void testObjectOverrides() throws UnknownHostException {
                                             .version(new ServerVersion(asList(2, 4, 1)))
                                             .minWireVersion(1)
                                             .maxWireVersion(2)
+                                            .electionId(new ObjectId())
                                             .exception(new IllegalArgumentException("This is illegal"));
         assertEquals(builder.build(), builder.build());
         assertEquals(builder.build().hashCode(), builder.build().hashCode());
@@ -389,4 +394,4 @@ public void serverWithMaxWireVersionLessThanDriverMinWireVersionShouldBeIncompat
 
     }
 
-}
+}
diff --git a/src/test/com/mongodb/ServerMonitorSpecification.groovy b/src/test/com/mongodb/ServerMonitorSpecification.groovy
@@ -1,4 +1,5 @@
 package com.mongodb
+
 import spock.lang.IgnoreIf
 
 import java.util.concurrent.CountDownLatch
@@ -45,6 +46,19 @@ class ServerMonitorSpecification extends FunctionalSpecification {
         newDescription.maxWriteBatchSize == expected
     }
 
+    def 'should set electionId'() {
+        given:
+        initializeServerMonitor(new ServerAddress())
+        CommandResult commandResult = database.command(new BasicDBObject('ismaster', 1))
+        def expected = commandResult.get('electionId')
+
+        when:
+        latch.await()
+
+        then:
+        newDescription.electionId == expected
+    }
+
     @IgnoreIf( { serverIsAtLeastVersion(2.6) } )
     def 'should set default max wire batch size when not provided by server'() {
         given:

Original file line number	Diff line number	Diff line change
`@@ -257,6 +257,7 @@ private ServerDescription createDescription(final CommandResult commandResult, f`
`257`	`257`	`.setName(commandResult.getString("setName"))`
`258`	`258`	`.minWireVersion(commandResult.getInt("minWireVersion", ServerDescription.getDefaultMinWireVersion()))`
`259`	`259`	`.maxWireVersion(commandResult.getInt("maxWireVersion", ServerDescription.getDefaultMaxWireVersion()))`
	`260`	`+ .electionId(commandResult.containsKey("electionId") ? commandResult.getObjectId("electionId") : null)`
`260`	`261`	`.averageLatency(averageLatencyNanos, TimeUnit.NANOSECONDS)`
`261`	`262`	`.ok(commandResult.ok()).build();`
`262`	`263`	`}`
`@@ -319,4 +320,4 @@ private ServerDescription getConnectingServerDescription(final Throwable throwab`
`319`	`320`	`.exception(throwable)`
`320`	`321`	`.build();`
`321`	`322`	`}`
`322`		`-}`
	`323`	`+}`