Fix DisableInitialHostLookup flag ignored when querying system.peers(PR update)

This commit provides update for the 'Modified DisableInitialHostLookup flag name and logic'
refreshRing() does NOOP when 'DisableInitialHostLookup' is true. system.peers
can be queried when flag is true for the schema version retrieving.

patch by Oleksandr Luzhniy; for CASSGO-45

Author: tengu-alt

CHANGELOG.md

@@ -50,6 +50,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Refactoring hostpool package test and Expose HostInfo creation (CASSGO-59)
 
+- Fix DisableInitialHostLookup flag ignored when querying system.peers (CASSGO-5)
+
+- Fix DisableInitialHostLookup flag ignored when querying system.peers (PR update) (CASSGO-5)
+
 ### Fixed
 - Cassandra version unmarshal fix (CASSGO-49)
 

cluster.go

@@ -195,12 +195,12 @@ type ClusterConfig struct {
 	// set to 10.0.0.1 which is what will be used to connect to.
 	IgnorePeerAddr bool
 
-	// If DisableHostLookup then the driver will not attempt to get host info
+	// If DisableInitialHostLookup then the driver will not attempt to get host info
 	// from the system.peers table, this will mean that the driver will connect to
 	// hosts supplied and will not attempt to lookup the hosts information, this will
 	// mean that data_center, rack and token information will not be available and as
 	// such host filtering and token aware query routing will not be available.
-	DisableHostLookup bool
+	DisableInitialHostLookup bool
 
 	// Configure events the driver will register for
 	Events struct {

conn.go

@@ -1904,14 +1904,12 @@ func (c *Conn) awaitSchemaAgreement(ctx context.Context) (err error) {
 	endDeadline := time.Now().Add(c.session.cfg.MaxWaitSchemaAgreement)
 
 	for time.Now().Before(endDeadline) {
-		iter := &Iter{}
-		if !c.session.cfg.DisableHostLookup {
-			iter = c.querySystemPeers(ctx, c.host.version)
+		iter := c.querySystemPeers(ctx, c.host.version)
 
-			rows, err := iter.SliceMap()
-			if err != nil {
-				goto cont
-			}
+		rows, err := iter.SliceMap()
+		if err != nil {
+			goto cont
+		}
 
 		for _, row := range rows {
 			h, err := NewHostInfo(c.host.ConnectAddress(), c.session.cfg.Port)
@@ -1927,12 +1925,11 @@ func (c *Conn) awaitSchemaAgreement(ctx context.Context) (err error) {
 				continue
 			}
 
-				versions[host.schemaVersion] = struct{}{}
-			}
+			versions[host.schemaVersion] = struct{}{}
+		}
 
-			if err = iter.Close(); err != nil {
-				goto cont
-			}
+		if err = iter.Close(); err != nil {
+			goto cont
 		}
 
 		iter = c.query(ctx, localSchemas)

disable_host_lookup_test.go

@@ -1,84 +1,39 @@
 package gocql
 
 import (
-	"fmt"
-	"log"
+	"net"
 	"testing"
-	"time"
 )
 
-func TestDisableHostLookup(t *testing.T) {
+func TestDisableHostLookupRingRefresh(t *testing.T) {
 	cluster := createCluster()
-	cluster.DisableHostLookup = true
+	cluster.DisableInitialHostLookup = true
 
 	cluster.NumConns = 1
 	session := createSessionFromCluster(cluster, t)
 	defer session.Close()
 
-	if err := createTable(session, "CREATE TABLE IF NOT EXISTS gocql_test.test_table (id int primary key)"); err != nil {
-		t.Fatal(err)
-	}
-	if err := session.Query("insert into gocql_test.test_table (id) values (?)", 123).Exec(); err != nil {
-		t.Error(err)
-		return
-	}
-
-	go simulateHeartbeatFailure(session)
-
-	queryCycles := 20
-	for ; queryCycles > 0; queryCycles-- {
-		time.Sleep(2 * time.Second)
-		queryTestKeyspace(session)
-		triggerSchemaChange(session)
-	}
-}
-
-func queryTestKeyspace(session *Session) {
-	iter := session.Query("SELECT * FROM gocql_test.test_table").Iter()
-
-	var id string
-	for iter.Scan(&id) {
-		fmt.Printf("id: %s\n", id)
-	}
-	if err := iter.Close(); err != nil {
-		log.Printf("error querying: %v", err)
-	}
-}
-
-func triggerSchemaChange(session *Session) {
-	// Create a keyspace to trigger schema agreement
-	err := session.Query(`CREATE KEYSPACE IF NOT EXISTS test_keyspace WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 1}`).Exec()
-	if err != nil {
-		log.Printf("unable to create keyspace: %v", err)
-	}
+	oldHosts := make(map[string]*HostInfo)
 
-	// Alter the keyspace to trigger schema agreement
-	err = session.Query(`ALTER KEYSPACE test_keyspace WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 2}`).Exec()
-	if err != nil {
-		log.Printf("unable to alter keyspace: %v", err)
+	for key, host := range session.ring.hosts {
+		host.broadcastAddress = net.ParseIP("10.10.10.10")
+		oldHosts[key] = host
 	}
 
-	// Drop the keyspace after schema agreement
-	err = session.Query(`DROP KEYSPACE IF EXISTS test_keyspace`).Exec()
+	// if DisableInitialHostLookup == true - host.broadcastAddress must not be updated.
+	err := session.refreshRing()
 	if err != nil {
-		log.Printf("unable to drop keyspace: %v", err)
+		t.Fatal(err)
 	}
-}
 
-func simulateHeartbeatFailure(session *Session) {
-	time.Sleep(20 * time.Second) // simulate 20 seconds of normal operation
-	log.Println("Simulating heartbeat failure...")
-	cluster := session.cfg
-	session.Close() // close the session to simulate a heartbeat failure
+	for key, host := range session.ring.hosts {
+		oldHost, ok := oldHosts[key]
+		if !ok {
+			t.Fatalf("old host not found for key: %s", key)
+		}
 
-	time.Sleep(10 * time.Second) // wait for 10 seconds to simulate downtime
-
-	// Reconnect
-	log.Println("Reconnecting...")
-	newSession, err := cluster.CreateSession()
-	if err != nil {
-		log.Fatalf("unable to create session: %v", err)
+		if !oldHost.broadcastAddress.Equal(host.broadcastAddress) {
+			t.Fatalf("broadcast addresses do not match for key: %s", key)
+		}
 	}
-	*session = *newSession
-	log.Println("Reconnected")
 }

host_source.go

@@ -665,11 +665,6 @@ func (r *ringDescriber) getClusterPeerInfo(localHost *HostInfo) ([]*HostInfo, er
 		return nil, errNoControl
 	}
 
-	// Check if host lookup is disabled
-	if r.session.cfg.DisableHostLookup {
-		return []*HostInfo{}, nil
-	}
-
 	var peers []*HostInfo
 	iter := r.session.control.withConnHost(func(ch *connHost) *Iter {
 		return ch.conn.querySystemPeers(context.TODO(), localHost.version)
@@ -752,6 +747,9 @@ func (s *Session) refreshRing() error {
 }
 
 func refreshRing(r *ringDescriber) error {
+	if r.session.cfg.DisableInitialHostLookup {
+		return nil
+	}
 	hosts, partitioner, err := r.GetHosts()
 	if err != nil {
 		return err

session.go

@@ -241,7 +241,7 @@ func (s *Session) init() error {
 			return err
 		}
 
-		if !s.cfg.DisableHostLookup {
+		if !s.cfg.DisableInitialHostLookup {
 			var partitioner string
 			newHosts, partitioner, err := s.hostSource.GetHosts()
 			if err != nil {
@@ -344,7 +344,7 @@ func (s *Session) init() error {
 	// cluster is using the newer system schema or not... however, if control
 	// connection is disable, we really have no choice, so we just make our
 	// best guess...
-	if !s.cfg.disableControlConn && s.cfg.DisableHostLookup {
+	if !s.cfg.disableControlConn && s.cfg.DisableInitialHostLookup {
 		newer, _ := checkSystemSchema(s.control)
 		s.useSystemSchema = newer
 	} else {