Add integration test for zero token nodes

Adds ZeroTokenNodesIT that checks the behaviour of the driver when zero-token
nodes are involved. Currently has one method `should_ignore_zero_token_peer`
in which driver expects a normal contact point and as a result it should not
populate the metadata with zero-token node from the peers table.

Author: Bouncheck

driver-core/src/test/java/com/datastax/driver/core/ZeroTokenNodesIT.java

@@ -0,0 +1,85 @@
+package com.datastax.driver.core;
+
+import static org.apache.log4j.Level.WARN;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.datastax.driver.core.utils.ScyllaVersion;
+import java.net.InetSocketAddress;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+public class ZeroTokenNodesIT {
+  private Logger logger = Logger.getLogger(ControlConnection.class);
+  private MemoryAppender appender;
+  private Level originalLevel;
+
+  @BeforeMethod(groups = "short")
+  public void startCapturingLogs() {
+    originalLevel = logger.getLevel();
+    logger.setLevel(WARN);
+    logger.addAppender(appender = new MemoryAppender());
+  }
+
+  @AfterMethod(groups = "short")
+  public void stopCapturingLogs() {
+    logger.setLevel(originalLevel);
+    logger.removeAppender(appender);
+  }
+
+  @Test(groups = "short")
+  @ScyllaVersion(
+      minOSS = "6.2.0",
+      minEnterprise = "2024.2.2",
+      description = "Zero-token nodes introduced in scylladb/19684")
+  public void should_ignore_zero_token_peer() {
+    // Given 4 node cluster with 1 zero-token node and normal contact point,
+    // make sure that it's not included in the metadata.
+    // By extension, it won't be included in the query planning.
+    Cluster cluster = null;
+    Session session = null;
+    CCMBridge bridgeA = null;
+    try {
+      bridgeA =
+          CCMBridge.builder().withNodes(3).withIpPrefix("127.0.1.").withBinaryPort(9042).build();
+      bridgeA.start();
+      bridgeA.add(4);
+      bridgeA.updateNodeConfig(4, "join_ring", false);
+      bridgeA.start(4);
+      bridgeA.waitForUp(4);
+
+      cluster =
+          Cluster.builder()
+              .addContactPointsWithPorts(new InetSocketAddress(bridgeA.ipOfNode(1), 9042))
+              .withPort(9042)
+              .withoutAdvancedShardAwareness()
+              .build();
+      session = cluster.connect();
+
+      ResultSet rs = session.execute("select * from system.local");
+      Row row = rs.one();
+      String address = row.getInet("broadcast_address").toString();
+      assertThat(address).endsWith(bridgeA.ipOfNode(1));
+      String line = null;
+      try {
+        line = appender.waitAndGet(5000);
+      } catch (InterruptedException e) {
+        throw new RuntimeException(e);
+      }
+      assertThat(line).contains("Found invalid row in system.peers");
+      assertThat(line).contains("tokens=null");
+      Set<Host> hosts = cluster.getMetadata().getAllHosts();
+      Set<String> toStrings = hosts.stream().map(Host::toString).collect(Collectors.toSet());
+      assertThat(toStrings).containsOnly("/127.0.1.1:9042", "/127.0.1.2:9042", "/127.0.1.3:9042");
+    } finally {
+      assert bridgeA != null;
+      bridgeA.close();
+      if (session != null) session.close();
+      if (cluster != null) cluster.close();
+    }
+  }
+}