JAVA-852: Ignore peers with null entries during auto-discovery.

Kevin Gallardo · olim7t · commit 4c75b91a1bad · 2016-03-10T20:46:01.000+01:00
diff --git a/changelog/README.md b/changelog/README.md
@@ -38,6 +38,7 @@
 - [improvement] JAVA-444: Add Java process information to UUIDs.makeNode() hash.
 - [improvement] JAVA-977: Preserve original cause when BuiltStatement value can't be serialized.
 - [bug] JAVA-1094: Backport TypeCodec parse and format fixes from 3.0.
+- [improvement] JAVA-852: Ignore peers with null entries during discovery.
 
 Merged from 2.0 branch:
 
diff --git a/driver-core/src/main/java/com/datastax/driver/core/ControlConnection.java b/driver-core/src/main/java/com/datastax/driver/core/ControlConnection.java
@@ -38,6 +38,8 @@ class ControlConnection implements Host.StateListener, Connection.Owner {
 
     private static final Logger logger = LoggerFactory.getLogger(ControlConnection.class);
 
+    private static final boolean EXTENDED_PEER_CHECK = SystemProperties.getBoolean("com.datastax.driver.EXTENDED_PEER_CHECK", true);
+
     private static final InetAddress bindAllAddress;
 
     static {
@@ -280,7 +282,7 @@ private Connection tryConnect(Host host, boolean isInitialConnection) throws Con
             // We need to refresh the node list again;
             // We want that because the token map was not properly initialized by the first call above,
             // since it requires the list of keyspaces to be loaded.
-            refreshNodeListAndTokenMap(connection, cluster, false, true);
+            refreshNodeListAndTokenMap(connection, cluster, false, false);
 
             return connection;
         } catch (BusyConnectionException e) {
@@ -421,19 +423,18 @@ void refreshNodeListAndTokenMap() {
         }
     }
 
-    private static InetSocketAddress addressToUseForPeerHost(Row peersRow, InetSocketAddress connectedHost, Cluster.Manager cluster, boolean logMissingRpcAddresses) {
+    private static InetSocketAddress addressToUseForPeerHost(Row peersRow, InetSocketAddress connectedHost, Cluster.Manager cluster) {
         InetAddress peer = peersRow.getInet("peer");
         InetAddress addr = peersRow.getInet("rpc_address");
 
-        if (peer.equals(connectedHost.getAddress()) || (addr != null && addr.equals(connectedHost.getAddress()))) {
+        // We've already called isValid on the row, which checks this
+        assert addr != null;
+
+        if (peer.equals(connectedHost.getAddress()) || addr.equals(connectedHost.getAddress())) {
             // Some DSE versions were inserting a line for the local node in peers (with mostly null values). This has been fixed, but if we
             // detect that's the case, ignore it as it's not really a big deal.
             logger.debug("System.peers on node {} has a line for itself. This is not normal but is a known problem of some DSE version. Ignoring the entry.", connectedHost);
             return null;
-        } else if (addr == null) {
-            if (logMissingRpcAddresses)
-                logger.warn("No rpc_address found for host {} in {}'s peers system table. {} will be ignored.", peer, connectedHost, peer);
-            return null;
         } else if (addr.equals(bindAllAddress)) {
             logger.warn("Found host with 0.0.0.0 as rpc_address, using listen_address ({}) to contact it instead. If this is incorrect you should avoid the use of 0.0.0.0 server side.", peer);
             addr = peer;
@@ -462,7 +463,7 @@ private Row fetchNodeInfo(Host host, Connection c) throws ConnectionException, B
         DefaultResultSetFuture future = new DefaultResultSetFuture(null, cluster.protocolVersion(), new Requests.Query(SELECT_PEERS));
         c.write(future);
         for (Row row : future.get()) {
-            InetSocketAddress addr = addressToUseForPeerHost(row, c.address, cluster, true);
+            InetSocketAddress addr = addressToUseForPeerHost(row, c.address, cluster);
             if (addr != null && addr.equals(host.getSocketAddress()))
                 return row;
         }
@@ -492,11 +493,13 @@ boolean refreshNodeInfo(Host host) {
                     logger.warn("No row found for host {} in {}'s peers system table. {} will be ignored.", host.getAddress(), c.address, host.getAddress());
                     return false;
                 }
-                // Ignore hosts with a null rpc_address, as this is most likely a phantom row in system.peers (JAVA-428).
-                // Don't test this for the control host since we're already connected to it anyway, and we read the info from system.local
-                // which doesn't have an rpc_address column (JAVA-546).
-            } else if (!c.address.equals(host.getSocketAddress()) && row.getInet("rpc_address") == null) {
-                logger.warn("No rpc_address found for host {} in {}'s peers system table. {} will be ignored.", host.getAddress(), c.address, host.getAddress());
+            }
+
+            // Ignore rows with invalid values, as this is most likely a phantom row in system.peers (JAVA-428,
+            // JAVA-852).
+            // Skip the control host since we're already connected to it anyway, and we read the info from system.local,
+            // which doesn't have an rpc_address column (JAVA-546).
+            if (!c.address.equals(host.getSocketAddress()) && !isValidPeer(row, true)) {
                 return false;
             }
 
@@ -554,7 +557,7 @@ private static void updateLocationInfo(Host host, String datacenter, String rack
             cluster.loadBalancingPolicy().onAdd(host);
     }
 
-    private static void refreshNodeListAndTokenMap(Connection connection, Cluster.Manager cluster, boolean isInitialConnection, boolean logMissingRpcAddresses) throws ConnectionException, BusyConnectionException, ExecutionException, InterruptedException {
+    private static void refreshNodeListAndTokenMap(Connection connection, Cluster.Manager cluster, boolean isInitialConnection, boolean logInvalidPeers) throws ConnectionException, BusyConnectionException, ExecutionException, InterruptedException {
         logger.debug("[Control connection] Refreshing node list and token map");
 
         boolean metadataEnabled = cluster.configuration.getQueryOptions().isMetadataEnabled();
@@ -603,10 +606,10 @@ private static void refreshNodeListAndTokenMap(Connection connection, Cluster.Ma
         List<Set<String>> allTokens = new ArrayList<Set<String>>();
 
         for (Row row : peersFuture.get()) {
-            InetSocketAddress addr = addressToUseForPeerHost(row, connection.address, cluster, logMissingRpcAddresses);
-            if (addr == null)
+            if (!isValidPeer(row, logInvalidPeers))
                 continue;
 
+            InetSocketAddress addr = addressToUseForPeerHost(row, connection.address, cluster);
             foundHosts.add(addr);
             dcs.add(row.getString("data_center"));
             racks.add(row.getString("rack"));
@@ -654,6 +657,46 @@ private static void refreshNodeListAndTokenMap(Connection connection, Cluster.Ma
             cluster.metadata.rebuildTokenMap(partitioner, tokenMap);
     }
 
+    private static boolean isValidPeer(Row peerRow, boolean logIfInvalid) {
+        boolean isValid = peerRow.getColumnDefinitions().contains("rpc_address")
+                && !peerRow.isNull("rpc_address");
+        if (EXTENDED_PEER_CHECK) {
+            isValid &= peerRow.getColumnDefinitions().contains("host_id")
+                    && !peerRow.isNull("host_id")
+                    && peerRow.getColumnDefinitions().contains("data_center")
+                    && !peerRow.isNull("data_center")
+                    && peerRow.getColumnDefinitions().contains("rack")
+                    && !peerRow.isNull("rack")
+                    && peerRow.getColumnDefinitions().contains("tokens")
+                    && !peerRow.isNull("tokens");
+        }
+        if (!isValid && logIfInvalid)
+            logger.warn("Found invalid row in system.peers: {}. " +
+                    "This is likely a gossip or snitch issue, this host will be ignored.", formatInvalidPeer(peerRow));
+        return isValid;
+    }
+
+    // Custom formatting to avoid spamming the logs if 'tokens' is present and contains a gazillion tokens
+    private static String formatInvalidPeer(Row peerRow) {
+        StringBuilder sb = new StringBuilder("[peer=" + peerRow.getInet("peer"));
+        formatMissingOrNullColumn(peerRow, "rpc_address", sb);
+        if (EXTENDED_PEER_CHECK) {
+            formatMissingOrNullColumn(peerRow, "host_id", sb);
+            formatMissingOrNullColumn(peerRow, "data_center", sb);
+            formatMissingOrNullColumn(peerRow, "rack", sb);
+            formatMissingOrNullColumn(peerRow, "tokens", sb);
+        }
+        sb.append("]");
+        return sb.toString();
+    }
+
+    private static void formatMissingOrNullColumn(Row peerRow, String columnName, StringBuilder sb) {
+        if (!peerRow.getColumnDefinitions().contains(columnName))
+            sb.append(", missing ").append(columnName);
+        else if (peerRow.isNull(columnName))
+            sb.append(", ").append(columnName).append("=null");
+    }
+
     boolean waitForSchemaAgreement() throws ConnectionException, BusyConnectionException, ExecutionException, InterruptedException {
         long start = System.nanoTime();
         long elapsed = 0;
@@ -690,7 +733,10 @@ boolean checkSchemaAgreement() throws ConnectionException, BusyConnectionExcepti
 
         for (Row row : peersFuture.get()) {
 
-            InetSocketAddress addr = addressToUseForPeerHost(row, connection.address, cluster, true);
+            if (!isValidPeer(row, false))
+                continue;
+
+            InetSocketAddress addr = addressToUseForPeerHost(row, connection.address, cluster);
             if (addr == null || row.isNull("schema_version"))
                 continue;
 
diff --git a/driver-core/src/test/java/com/datastax/driver/core/ClusterInitTest.java b/driver-core/src/test/java/com/datastax/driver/core/ClusterInitTest.java
@@ -35,6 +35,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
@@ -140,8 +141,7 @@ public void should_handle_failing_or_missing_contact_points() throws UnknownHost
                 }
                 assertThat(cluster).host(failingHost.address).isReconnectingFromDown();
             }
-            assertThat(cluster).host(missingHostAddress).isNull();
-
+            assertThat(TestUtils.findHost(cluster, missingHostAddress)).isNull();
         } finally {
             if (cluster != null)
                 cluster.close();
@@ -275,6 +275,7 @@ private void primePeerRows(Scassandra scassandra, List<FakeHost> otherHosts) thr
                     .put("rack", "rack1")
                     .put("release_version", "2.0.1")
                     .put("tokens", ImmutableSet.of(Long.toString(Long.MIN_VALUE + i++)))
+                    .put("host_id", UUID.randomUUID())
                     .build());
         }
 
diff --git a/driver-core/src/test/java/com/datastax/driver/core/ScassandraCluster.java b/driver-core/src/test/java/com/datastax/driver/core/ScassandraCluster.java
@@ -17,6 +17,7 @@
 
 import com.beust.jcommander.internal.Lists;
 import com.beust.jcommander.internal.Maps;
+import com.datastax.driver.core.utils.UUIDs;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
@@ -310,6 +311,7 @@ private void primeMetadata(Scassandra node) {
                             .put("rack", "rack1")
                             .put("release_version", getPeerInfo(dc, n + 1, "release_version", "2.1.8"))
                             .put("tokens", ImmutableSet.of(Long.toString(tokens.get(n))))
+                            .put("host_id", UUIDs.random())
                             .build();
                     rows.add(row);
                 }
@@ -373,7 +375,8 @@ private Object getPeerInfo(int dc, int node, String property, Object defaultValu
             column("data_center", TEXT),
             column("rack", TEXT),
             column("release_version", TEXT),
-            column("tokens", set(TEXT))
+            column("tokens", set(TEXT)),
+            column("host_id", UUID)
     };
 
     public static final org.scassandra.http.client.types.ColumnMetadata[] SELECT_LOCAL = {
