Prevent syntetic CREATED events on HostSelectionPolicy.KeyspaceChanged during session init

This patch fixes regression introduced in CASSGO-104 that caused the driver to fire syntetic
CREATED events on HostSelectionPolicy.KeyspaceChanged for each existing keyspace during session initialization.

Author: worryg0d

cluster.go

@@ -409,6 +409,8 @@ func (cfg *ClusterConfig) filterHost(host *HostInfo) bool {
 type MetadataConfig struct {
 	// CacheMode controls how the driver reads and caches schema metadata from Cassandra system tables.
 	//
+	// It is required to be enabled for [TokenAwareHostPolicy] and custom host selection policies.
+	//
 	// Also, it affects the behavior of schema change listeners.
 	//
 	// If CacheMode is [KeyspaceOnly], only [KeyspaceChangeListener] will be notified,

common_test.go

@@ -303,6 +303,10 @@ func randomText(size int) string {
 	return string(result)
 }
 
+func randomNameWithPrefix(prefix string) string {
+	return prefix + strings.ToLower(randomText(10))
+}
+
 func assertEqual(t *testing.T, description string, expected, actual interface{}) {
 	t.Helper()
 	if expected != actual {

integration_test.go

@@ -35,9 +35,11 @@ import (
 	"net"
 	"reflect"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
 	inf "gopkg.in/inf.v0"
 )
 
@@ -1028,3 +1030,87 @@ func TestSmallTimeoutNoPoolErrors(t *testing.T) {
 			errorCount, logOutput)
 	}
 }
+
+// Emulates round-robin host selection policy and captures KeyspaceChanged events.
+type capturingRRTestPolicy struct {
+	roundRobinHostPolicy
+
+	capturedEvents []KeyspaceUpdateEvent
+	mu             sync.Mutex
+}
+
+func (c *capturingRRTestPolicy) KeyspaceChanged(event KeyspaceUpdateEvent) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.capturedEvents = append(c.capturedEvents, event)
+	c.roundRobinHostPolicy.KeyspaceChanged(event)
+}
+
+func (c *capturingRRTestPolicy) CapturedEvents() []KeyspaceUpdateEvent {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return append([]KeyspaceUpdateEvent{}, c.capturedEvents...)
+}
+
+func (c *capturingRRTestPolicy) ResetEvents() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.capturedEvents = nil
+}
+
+func TestHostSelectionPolicyKeyspaceChangedEvents(t *testing.T) {
+	cluster := createCluster()
+	policy := capturingRRTestPolicy{}
+	cluster.Metadata.CacheMode = Full
+	cluster.PoolConfig.HostSelectionPolicy = &policy
+
+	session, err := cluster.CreateSession()
+	require.NoError(t, err)
+	defer session.Close()
+
+	// Regression test for CASSGO-132: KeyspaceChanged events are fired during session creation.
+	events := policy.CapturedEvents()
+	require.Empty(t, events, "KeyspaceChanged events should not be fired during session creation")
+
+	// Check keyspace events are fired
+
+	// CREATE
+	ksName := randomNameWithPrefix("gocql_test_")
+	err = session.Query(fmt.Sprintf("CREATE KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}", ksName)).ExecContext(context.Background())
+	require.NoError(t, err)
+
+	require.Eventually(t, func() bool {
+		events = policy.CapturedEvents()
+		return len(events) == 1
+	}, time.Second*5, time.Millisecond*100)
+
+	require.Len(t, events, 1)
+	require.Equal(t, ksName, events[0].Keyspace)
+	require.Equal(t, SchemaChangeTypeCreated, events[0].Change)
+
+	// ALTER
+	policy.ResetEvents()
+	err = session.Query(fmt.Sprintf("ALTER KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '2'}", ksName)).ExecContext(context.Background())
+	require.NoError(t, err)
+	require.Eventually(t, func() bool {
+		events = policy.CapturedEvents()
+		return len(events) == 1
+	}, time.Second*5, time.Millisecond*100)
+
+	require.Len(t, events, 1)
+	require.Equal(t, ksName, events[0].Keyspace)
+	require.Equal(t, SchemaChangeTypeUpdated, events[0].Change)
+
+	// DROP
+	policy.ResetEvents()
+	err = session.Query(fmt.Sprintf("DROP KEYSPACE %s", ksName)).ExecContext(context.Background())
+	require.NoError(t, err)
+	require.Eventually(t, func() bool {
+		events = policy.CapturedEvents()
+		return len(events) == 1
+	}, time.Second*5, time.Millisecond*100)
+
+	require.Len(t, events, 1)
+	require.Equal(t, ksName, events[0].Keyspace)
+	require.Equal(t, SchemaChangeTypeDropped, events[0].Change)
+}

metadata.go

@@ -575,6 +575,10 @@ type schemaDescriber struct {
 // Schema change events are server-initiated messages sent to clients that have registered
 // for schema change notifications. These events indicate modifications to keyspaces, tables,
 // user-defined types, functions, or aggregates.
+//
+// When a schema change event is received, the driver will update its internal metadata
+// by comparing the old and new metadata. If the metadata has changed, the driver will
+// notify [HostSelectionPolicy.KeyspaceChanged] method about the change.
 const (
 	SchemaChangeTypeCreated = "CREATED" // Schema object was created
 	SchemaChangeTypeUpdated = "UPDATED" // Schema object was modified
@@ -858,9 +862,6 @@ func refreshSchemas(session *Session) error {
 
 	// Notify policy if it supports schema refresh notifications
 	notifier, supportsRefresh := session.policy.(schemaRefreshNotifier)
-	hasKeyspaceListener := session.schemaListeners.hasKeyspace() && sessionInitialized
-
-	// Notify the policy if it supports schema refresh notifications
 	if supportsRefresh {
 		notifier.schemaRefreshed(session.schemaDescriber.getSchemaMetaForRead())
 	}
@@ -871,31 +872,36 @@ func refreshSchemas(session *Session) error {
 		NewLogFieldInt("dropped_keyspaces_count", len(droppedKeyspaces)),
 	)
 
-	// If we don't support schemaRefreshed OR we have listeners, we need to loop
-	if !supportsRefresh || hasKeyspaceListener {
+	// We should not call KeyspaceChanged method if policy implements schemaRefreshNotifier.
+	shouldCallKeyspaceChanged := !supportsRefresh && sessionInitialized
+	// We should not notify keyspace listeners if they are not set or session is not initialized yet.
+	shouldNotifyKeyspaceListener := session.schemaListeners.hasKeyspace() && sessionInitialized
+
+	// Loop only if we need to notify policy or keyspace listeners.
+	if shouldCallKeyspaceChanged || shouldNotifyKeyspaceListener {
 		for _, name := range newKeyspaces {
-			if !supportsRefresh {
+			if shouldCallKeyspaceChanged {
 				session.policy.KeyspaceChanged(KeyspaceUpdateEvent{Keyspace: name, Change: SchemaChangeTypeCreated})
 			}
-			if hasKeyspaceListener {
+			if shouldNotifyKeyspaceListener {
 				session.schemaListeners.OnKeyspaceCreated(OnKeyspaceCreatedEvent{Keyspace: keyspaces[name].Clone()})
 			}
 		}
 
 		for _, name := range droppedKeyspaces {
-			if !supportsRefresh {
+			if shouldCallKeyspaceChanged {
 				session.policy.KeyspaceChanged(KeyspaceUpdateEvent{Keyspace: name, Change: SchemaChangeTypeDropped})
 			}
-			if hasKeyspaceListener {
+			if shouldNotifyKeyspaceListener {
 				session.schemaListeners.OnKeyspaceDropped(OnKeyspaceDroppedEvent{Keyspace: oldKeyspaceMeta[name].Clone()})
 			}
 		}
 
 		for _, name := range updatedKeyspaces {
-			if !supportsRefresh {
+			if shouldCallKeyspaceChanged {
 				session.policy.KeyspaceChanged(KeyspaceUpdateEvent{Keyspace: name, Change: SchemaChangeTypeUpdated})
 			}
-			if hasKeyspaceListener {
+			if shouldNotifyKeyspaceListener {
 				session.schemaListeners.OnKeyspaceUpdated(OnKeyspaceUpdatedEvent{
 					Old: oldKeyspaceMeta[name].Clone(),
 					New: keyspaces[name].Clone(),

policies.go

@@ -427,6 +427,9 @@ func NonLocalReplicasFallback() func(policy *tokenAwareHostPolicy) {
 }
 
 // ShuffledTokenAwareHostPolicy is a token aware host selection policy that shuffles replicas.
+//
+// Note: TokenAwareHostPolicy requires the metadata cache to be enabled.
+// See [ClusterConfig.Metadata] for more details.
 func ShuffledTokenAwareHostPolicy(fallback HostSelectionPolicy, opts ...func(*tokenAwareHostPolicy)) HostSelectionPolicy {
 	p := &tokenAwareHostPolicy{
 		fallback:                fallback,
@@ -442,6 +445,9 @@ func ShuffledTokenAwareHostPolicy(fallback HostSelectionPolicy, opts ...func(*to
 // TokenAwareHostPolicy is a token aware host selection policy, where hosts are
 // selected based on the partition key, so queries are sent to the host which
 // owns the partition. Fallback is used when routing information is not available.
+//
+// Note: TokenAwareHostPolicy requires the metadata cache to be enabled.
+// See [ClusterConfig.Metadata] for more details.
 func TokenAwareHostPolicy(fallback HostSelectionPolicy, opts ...func(*tokenAwareHostPolicy)) HostSelectionPolicy {
 	p := &tokenAwareHostPolicy{fallback: fallback}
 	for _, opt := range opts {
@@ -1080,3 +1086,16 @@ type SimpleSpeculativeExecution struct {
 
 func (sp *SimpleSpeculativeExecution) Attempts() int        { return sp.NumAttempts }
 func (sp *SimpleSpeculativeExecution) Delay() time.Duration { return sp.TimeoutDelay }
+
+// hostSelectionPolicyRequiresMetadata returns true if the host selection policy requires the metadata cache to be enabled.
+// [tokenAwareHostPolicy] is the only builtin policy that requires the metadata cache to be enabled.
+// Custom policies also require the metadata cache to be enabled.
+func hostSelectionPolicyRequiresMetadata(p HostSelectionPolicy) bool {
+	switch p.(type) {
+	case *rackAwareRR, *roundRobinHostPolicy, *singleHostReadyPolicy, *dcAwareRR:
+		return false
+	default:
+		// tokenAwareHostPolicy and custom policies are required to have the metadata cache enabled.
+		return true
+	}
+}

policies_test.go

@@ -37,6 +37,8 @@ import (
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/stretchr/testify/assert"
 )
 
 // Tests of the round-robin host selection policy implementation
@@ -1099,3 +1101,14 @@ func TestHostPolicy_TokenAware_TopologyChangeUpdatesAllKeyspaces(t *testing.T) {
 		}
 	})
 }
+
+func Test_hostSelectionPolicyRequiresMetadataCache(t *testing.T) {
+	// Doesn't require metadata cache
+	assert.False(t, hostSelectionPolicyRequiresMetadata(RoundRobinHostPolicy()))
+	assert.False(t, hostSelectionPolicyRequiresMetadata(DCAwareRoundRobinPolicy("dc1")))
+	assert.False(t, hostSelectionPolicyRequiresMetadata(RackAwareRoundRobinPolicy("dc1", "rack1")))
+	assert.False(t, hostSelectionPolicyRequiresMetadata(SingleHostReadyPolicy(RoundRobinHostPolicy())))
+
+	// Requires metadata cache
+	assert.True(t, hostSelectionPolicyRequiresMetadata(TokenAwareHostPolicy(RoundRobinHostPolicy())))
+}

schema_events_test.go

@@ -23,7 +23,6 @@ package gocql
 
 import (
 	"fmt"
-	"strings"
 	"testing"
 	"time"
 
@@ -263,10 +262,6 @@ func testSchemaEventsKeyspace(t *testing.T, session *Session, listener *schemaCh
 	require.Equal(t, ks, listener.KeyspaceDroppedEvents[0].Keyspace.Name, "Expected keyspace dropped event to have correct keyspace name")
 }
 
-func randomNameWithPrefix(prefix string) string {
-	return prefix + strings.ToLower(randomText(10))
-}
-
 func testSchemaEventsTable(t *testing.T, session *Session, listener *schemaChangesTestListener) {
 	table := randomNameWithPrefix("gocql_table_events_")
 

session.go

@@ -149,6 +149,14 @@ func NewSession(cfg ClusterConfig) (*Session, error) {
 		return nil, fmt.Errorf("the default SerialConsistency level is not allowed to be anything else but SERIAL or LOCAL_SERIAL. Recived value: %v", cfg.SerialConsistency)
 	}
 
+	if cfg.PoolConfig.HostSelectionPolicy == nil {
+		cfg.PoolConfig.HostSelectionPolicy = RoundRobinHostPolicy()
+	}
+
+	if cfg.Metadata.CacheMode == Disabled && hostSelectionPolicyRequiresMetadata(cfg.PoolConfig.HostSelectionPolicy) {
+		return nil, ErrMetadataCacheRequired
+	}
+
 	// TODO: we should take a context in here at some point
 	ctx, cancel := context.WithCancel(context.TODO())
 
@@ -222,9 +230,6 @@ func NewSession(cfg ClusterConfig) (*Session, error) {
 	}
 	s.connCfg = connCfg
 
-	if cfg.PoolConfig.HostSelectionPolicy == nil {
-		cfg.PoolConfig.HostSelectionPolicy = RoundRobinHostPolicy()
-	}
 	s.pool = cfg.PoolConfig.buildPool(s)
 	s.policy = cfg.PoolConfig.HostSelectionPolicy
 
@@ -2517,16 +2522,17 @@ func (e Error) Error() string {
 }
 
 var (
-	ErrNotFound             = errors.New("not found")
-	ErrUnavailable          = errors.New("unavailable")
-	ErrUnsupported          = errors.New("feature not supported")
-	ErrTooManyStmts         = errors.New("too many statements")
-	ErrUseStmt              = errors.New("use statements aren't supported. Please see https://github.com/apache/cassandra-gocql-driver for explanation.")
-	ErrSessionClosed        = errors.New("session has been closed")
-	ErrNoConnections        = errors.New("gocql: no hosts available in the pool")
-	ErrNoKeyspace           = errors.New("no keyspace provided")
-	ErrKeyspaceDoesNotExist = errors.New("keyspace does not exist")
-	ErrNoMetadata           = errors.New("no metadata available")
+	ErrNotFound              = errors.New("not found")
+	ErrUnavailable           = errors.New("unavailable")
+	ErrUnsupported           = errors.New("feature not supported")
+	ErrTooManyStmts          = errors.New("too many statements")
+	ErrUseStmt               = errors.New("use statements aren't supported. Please see https://github.com/apache/cassandra-gocql-driver for explanation.")
+	ErrSessionClosed         = errors.New("session has been closed")
+	ErrNoConnections         = errors.New("gocql: no hosts available in the pool")
+	ErrNoKeyspace            = errors.New("no keyspace provided")
+	ErrKeyspaceDoesNotExist  = errors.New("keyspace does not exist")
+	ErrNoMetadata            = errors.New("no metadata available")
+	ErrMetadataCacheRequired = errors.New("metadata cache must be enabled for the selected host selection policy")
 )
 
 // ErrProtocol represents a protocol-level error.