Remove from the cluster description a replica set member whose "me" hostname:port does not match the one that was used to connect to it.

 JAVA-1815

Author: jyemin

src/main/com/mongodb/MultiServerCluster.java

@@ -187,6 +187,11 @@ private boolean handleReplicaSetMemberChanged(final ServerDescription newDescrip
 
         ensureServers(newDescription);
 
+        if (newDescription.getCanonicalAddress() != null && !newDescription.getAddress().sameHost(newDescription.getCanonicalAddress())) {
+            removeServer(newDescription.getAddress());
+            return true;
+        }
+
         if (newDescription.isPrimary()) {
             if (newDescription.getElectionId() != null) {
                 if (maxElectionId != null && maxElectionId.compareTo(newDescription.getElectionId()) > 0) {
@@ -239,7 +244,10 @@ private void addServer(final ServerAddress serverAddress) {
     }
 
     private void removeServer(final ServerAddress serverAddress) {
-        addressToServerTupleMap.remove(serverAddress).server.close();
+        ServerTuple removed = addressToServerTupleMap.remove(serverAddress);
+        if (removed != null) {
+            removed.server.close();
+        }
     }
 
     private void invalidateOldPrimaries(final ServerAddress newPrimary) {

src/main/com/mongodb/ServerDescription.java

@@ -54,6 +54,7 @@ class ServerDescription {
     private final ServerAddress address;
 
     private final ServerType type;
+    private final String canonicalAddress;
     private final Set<String> hosts;
     private final Set<String> passives;
     private final Set<String> arbiters;
@@ -74,11 +75,12 @@ class ServerDescription {
 
     private final ObjectId electionId;
 
-    private Throwable exception;
+    private final Throwable exception;
 
     static class Builder {
         private ServerAddress address;
         private ServerType type = Unknown;
+        private String canonicalAddress;
         private Set<String> hosts = Collections.emptySet();
         private Set<String> passives = Collections.emptySet();
         private Set<String> arbiters = Collections.emptySet();
@@ -108,6 +110,11 @@ public Builder type(final ServerType type) {
             return this;
         }
 
+        public Builder canonicalAddress(final String canonicalAddress) {
+            this.canonicalAddress = canonicalAddress;
+            return this;
+        }
+
         public Builder hosts(final Set<String> hosts) {
             this.hosts = hosts == null ? Collections.<String>emptySet() : Collections.unmodifiableSet(new HashSet<String>(hosts));
             return this;
@@ -277,6 +284,16 @@ public boolean isSecondary() {
         return ok && (type == ReplicaSetSecondary || type == ShardRouter || type == StandAlone);
     }
 
+    /**
+     * Gets the string representing the host name and port that this member of a replica set was configured with,
+     * e.g. {@code "somehost:27019"}. This is typically derived from the "me" field from the "isMaster" command response.
+     *
+     * @return the host name and port that this replica set member is configured with.
+     */
+    public String getCanonicalAddress() {
+        return canonicalAddress;
+    }
+
     public Set<String> getHosts() {
         return hosts;
     }
@@ -411,6 +428,9 @@ public boolean equals(final Object o) {
         if (!address.equals(that.address)) {
             return false;
         }
+        if (canonicalAddress != null ? !canonicalAddress.equals(that.canonicalAddress) : that.canonicalAddress != null) {
+            return false;
+        }
         if (!arbiters.equals(that.arbiters)) {
             return false;
         }
@@ -469,6 +489,7 @@ public boolean equals(final Object o) {
     public int hashCode() {
         int result = address.hashCode();
         result = 31 * result + type.hashCode();
+        result = 31 * result + (canonicalAddress != null ? canonicalAddress.hashCode() : 0);
         result = 31 * result + hosts.hashCode();
         result = 31 * result + passives.hashCode();
         result = 31 * result + arbiters.hashCode();
@@ -494,6 +515,7 @@ public String toString() {
         return "ServerDescription{"
                + "address=" + address
                + ", type=" + type
+               + ", canonicalAddress=" + canonicalAddress
                + ", hosts=" + hosts
                + ", passives=" + passives
                + ", arbiters=" + arbiters
@@ -550,6 +572,7 @@ private String getAverageLatencyFormattedInMilliseconds() {
         type = notNull("type", builder.type);
         state = notNull("state", builder.state);
         version = notNull("version", builder.version);
+        canonicalAddress = builder.canonicalAddress;
         hosts = builder.hosts;
         passives = builder.passives;
         arbiters = builder.arbiters;

src/main/com/mongodb/ServerMonitor.java

@@ -245,6 +245,7 @@ private ServerDescription createDescription(final CommandResult commandResult, f
                                 .version(serverVersion)
                                 .address(commandResult.getServerUsed())
                                 .type(getServerType(commandResult))
+                                .canonicalAddress(commandResult.getString("me"))
                                 .hosts(listToSet((List<String>) commandResult.get("hosts")))
                                 .passives(listToSet((List<String>) commandResult.get("passives")))
                                 .arbiters(listToSet((List<String>) commandResult.get("arbiters")))

src/test/com/mongodb/MultiServerClusterSpecification.groovy

@@ -115,6 +115,34 @@ class MultiServerClusterSpecification extends Specification {
         getClusterDescription(cluster).all == getServerDescriptions(firstServer, secondServer, thirdServer)
     }
 
+    def 'should remove a secondary server whose reported host name does not match the address connected to'() {
+        given:
+        def seedListAddress = new ServerAddress('127.0.0.1:27017')
+        def cluster = new MultiServerCluster(CLUSTER_ID,
+                                             ClusterSettings.builder().hosts([seedListAddress]).build(), factory,
+                                             CLUSTER_LISTENER);
+
+        when:
+        sendNotification(seedListAddress, ReplicaSetSecondary, [firstServer, secondServer], 'test', firstServer)
+
+        then:
+        getClusterDescription(cluster).all == getServerDescriptions(firstServer, secondServer)
+    }
+
+    def 'should remove a primary server whose reported host name does not match the address connected to'() {
+        given:
+        def seedListAddress = new ServerAddress('127.0.0.1:27017')
+        def cluster = new MultiServerCluster(CLUSTER_ID,
+                                             ClusterSettings.builder().hosts([seedListAddress]).build(), factory,
+                                             CLUSTER_LISTENER);
+
+        when:
+        sendNotification(seedListAddress, ReplicaSetPrimary, [firstServer, secondServer], 'test', firstServer)
+
+        then:
+        getClusterDescription(cluster).all == getServerDescriptions(firstServer, secondServer)
+    }
+
     def 'should remove a server when it no longer appears in hosts reported by the primary'() {
         given:
         def cluster = new MultiServerCluster(CLUSTER_ID,
@@ -480,19 +508,25 @@ class MultiServerClusterSpecification extends Specification {
         sendNotification(serverAddress, serverType, hosts, [], setName)
     }
 
+    def sendNotification(ServerAddress serverAddress, ServerType serverType, List<ServerAddress> hosts, String setName,
+                         ServerAddress trueAddress) {
+        factory.getServer(serverAddress).sendNotification(getBuilder(serverAddress, serverType, hosts, [], true, setName, null, trueAddress)
+                                                                  .build())
+    }
+
     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, null)
+        factory.getServer(serverAddress).sendNotification(getBuilder(serverAddress, serverType, hosts, passives, true, setName, null, null)
                                                                   .build())
     }
 
     def sendNotification(ServerAddress serverAddress, ServerType serverType, List<ServerAddress> hosts, ObjectId electionId) {
-        factory.getServer(serverAddress).sendNotification(getBuilder(serverAddress, serverType, hosts, [], true, 'test', electionId)
+        factory.getServer(serverAddress).sendNotification(getBuilder(serverAddress, serverType, hosts, [], true, 'test', electionId, null)
                                                                   .build())
     }
 
     def sendNotification(ServerAddress serverAddress, ServerType serverType, List<ServerAddress> hosts, boolean ok) {
-        factory.getServer(serverAddress).sendNotification(getBuilder(serverAddress, serverType, hosts, [], ok, null, null).build())
+        factory.getServer(serverAddress).sendNotification(getBuilder(serverAddress, serverType, hosts, [], ok, null, null, null).build())
     }
 
     def getClusterDescription(MultiServerCluster cluster) {
@@ -508,10 +542,11 @@ class MultiServerClusterSpecification extends Specification {
     }
 
     def getBuilder(ServerAddress serverAddress, ServerType serverType, List<ServerAddress> hosts, List<ServerAddress> passives, boolean ok,
-                   String setName, ObjectId electionId) {
+                   String setName, ObjectId electionId, ServerAddress trueAddress) {
         ServerDescription.builder()
                          .address(serverAddress)
                          .type(serverType)
+                         .canonicalAddress(trueAddress == null ? serverAddress.toString() : trueAddress.toString())
                          .ok(ok)
                          .state(Connected)
                          .hosts(hosts*.toString() as Set)

src/test/com/mongodb/ServerDescriptionTest.java

@@ -80,6 +80,7 @@ public void testDefaults() throws UnknownHostException {
         assertNull(serverDescription.getPrimary());
         assertEquals(Collections.<String>emptySet(), serverDescription.getHosts());
         assertEquals(new TagSet(), serverDescription.getTagSet());
+        assertNull(serverDescription.getCanonicalAddress());
         assertEquals(Collections.<String>emptySet(), serverDescription.getHosts());
         assertEquals(Collections.<String>emptySet(), serverDescription.getPassives());
         assertNull(serverDescription.getSetName());
@@ -103,6 +104,7 @@ public void testBuilder() throws UnknownHostException {
                                               .maxWriteBatchSize(1024)
                                               .averageLatency(50000, java.util.concurrent.TimeUnit.NANOSECONDS)
                                               .primary("localhost:27017")
+                                              .canonicalAddress("localhost:27018")
                                               .hosts(new HashSet<String>(asList("localhost:27017",
                                                                                 "localhost:27018",
                                                                                 "localhost:27019",
@@ -138,6 +140,7 @@ public void testBuilder() throws UnknownHostException {
         assertEquals(1024, serverDescription.getMaxWriteBatchSize());
 
         assertEquals("localhost:27017", serverDescription.getPrimary());
+        assertEquals("localhost:27018", serverDescription.getCanonicalAddress());
         assertEquals(new HashSet<String>(asList("localhost:27017", "localhost:27018", "localhost:27019", "localhost:27020")),
                      serverDescription.getHosts());
         assertEquals(new TagSet(new Tag("dc", "ny")), serverDescription.getTagSet());
@@ -163,6 +166,7 @@ public void testObjectOverrides() throws UnknownHostException {
                                             .maxWriteBatchSize(1024)
                                             .averageLatency(50000, java.util.concurrent.TimeUnit.NANOSECONDS)
                                             .primary("localhost:27017")
+                                            .canonicalAddress("localhost:27017")
                                             .hosts(new HashSet<String>(asList("localhost:27017",
                                                                               "localhost:27018")))
                                             .passives(new HashSet<String>(asList("localhost:27019")))

src/test/com/mongodb/ServerMonitorSpecification.groovy

@@ -32,6 +32,19 @@ class ServerMonitorSpecification extends FunctionalSpecification {
         newDescription.version == expectedVersion
     }
 
+    def 'should set canonicalAddress'() {
+        given:
+        initializeServerMonitor(new ServerAddress())
+        CommandResult commandResult = database.command(new BasicDBObject('ismaster', 1))
+        def expectedCanonicalAddress = commandResult.getString('me')
+
+        when:
+        latch.await()
+
+        then:
+        newDescription.canonicalAddress == expectedCanonicalAddress
+    }
+
     @IgnoreIf( { !serverIsAtLeastVersion(2.6) } )
     def 'should set max wire batch size when provided by server'() {
         given: