Merge pull request apache#592 from datastax/java852

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

Author: olim7t

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:
 

driver-core/pom.xml

@@ -15,7 +15,8 @@
       limitations under the License.
 
 -->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
   <modelVersion>4.0.0</modelVersion>
   <parent>
     <groupId>com.datastax.cassandra</groupId>
@@ -25,7 +26,9 @@
   <artifactId>cassandra-driver-core</artifactId>
   <packaging>bundle</packaging>
   <name>DataStax Java Driver for Apache Cassandra - Core</name>
-  <description>A driver for Apache Cassandra 1.2+ that works exclusively with the Cassandra Query Language version 3 (CQL3) and Cassandra's binary protocol.</description>
+  <description>A driver for Apache Cassandra 1.2+ that works exclusively with the Cassandra Query Language version 3
+    (CQL3) and Cassandra's binary protocol.
+  </description>
   <url>https://github.com/datastax/java-driver</url>
 
   <properties>
@@ -67,17 +70,17 @@
     <!-- Each of them is only a mandatory runtime dependency if you want to use the compression it offers -->
 
     <dependency>
-        <groupId>org.xerial.snappy</groupId>
-        <artifactId>snappy-java</artifactId>
-        <version>${snappy.version}</version>
-        <optional>true</optional>
+      <groupId>org.xerial.snappy</groupId>
+      <artifactId>snappy-java</artifactId>
+      <version>${snappy.version}</version>
+      <optional>true</optional>
     </dependency>
 
     <dependency>
-        <groupId>net.jpountz.lz4</groupId>
-        <artifactId>lz4</artifactId>
-        <version>${lz4.version}</version>
-        <optional>true</optional>
+      <groupId>net.jpountz.lz4</groupId>
+      <artifactId>lz4</artifactId>
+      <version>${lz4.version}</version>
+      <optional>true</optional>
     </dependency>
 
     <!-- End of compression libraries -->
@@ -221,7 +224,9 @@
         <executions>
           <execution>
             <phase>package</phase>
-            <goals><goal>shade</goal></goals>
+            <goals>
+              <goal>shade</goal>
+            </goals>
             <configuration>
               <shadedArtifactAttached>true</shadedArtifactAttached>
               <artifactSet>
@@ -287,7 +292,7 @@
                     </goals>
                   </pluginExecutionFilter>
                   <action>
-                    <ignore />
+                    <ignore/>
                   </action>
                 </pluginExecution>
               </pluginExecutions>
@@ -375,6 +380,7 @@
               <includes>
                 <include>**/SSL*Test.java</include>
                 <include>**/UUIDsPID*.java</include>
+                <include>**/ControlConnectionTest.java</include>
               </includes>
             </configuration>
           </plugin>

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;
 

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());
         }