[CDNC-4831] Added option to pass consistency level in the cassandra schema versio…

cmd/server/cadence/cadence.go

@@ -29,6 +29,8 @@ import (
 	"strings"
 	"syscall"
 
+	"github.com/uber/cadence/common/persistence/nosql/nosqlplugin/cassandra/gocql"
+
 	"github.com/urfave/cli"
 
 	"github.com/uber/cadence/common"
@@ -69,7 +71,7 @@ func startHandler(c *cli.Context) {
 		log.Fatalf("config validation failed: %v", err)
 	}
 	// cassandra schema version validation
-	if err := cassandra.VerifyCompatibleVersion(cfg.Persistence); err != nil {
+	if err := cassandra.VerifyCompatibleVersion(cfg.Persistence, gocql.All); err != nil {
 		log.Fatal("cassandra schema version compatibility check failed: ", err)
 	}
 	// sql schema version validation

cmd/server/cadence/server_test.go

@@ -29,6 +29,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/uber/cadence/common/persistence/nosql/nosqlplugin/cassandra/gocql"
+
 	"github.com/uber/cadence/testflags"
 
 	"github.com/stretchr/testify/require"
@@ -94,7 +96,7 @@ func (s *ServerSuite) TestServerStartup() {
 		log.Fatalf("config validation failed: %v", err)
 	}
 	// cassandra schema version validation
-	if err := cassandra.VerifyCompatibleVersion(cfg.Persistence); err != nil {
+	if err := cassandra.VerifyCompatibleVersion(cfg.Persistence, gocql.All); err != nil {
 		log.Fatal("cassandra schema version compatibility check failed: ", err)
 	}
 

common/persistence/nosql/nosqlplugin/cassandra/tests/cassandra_tool_version_test.go

@@ -28,6 +28,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/uber/cadence/common/persistence/nosql/nosqlplugin/cassandra/gocql"
+
 	log "github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
@@ -92,7 +94,7 @@ func (s *VersionTestSuite) TestVerifyCompatibleVersion() {
 		TransactionSizeLimit: dynamicconfig.GetIntPropertyFn(common.DefaultTransactionSizeLimit),
 		ErrorInjectionRate:   dynamicconfig.GetFloatPropertyFn(0),
 	}
-	s.NoError(cassandra.VerifyCompatibleVersion(cfg))
+	s.NoError(cassandra.VerifyCompatibleVersion(cfg, gocql.All))
 }
 
 func (s *VersionTestSuite) TestCheckCompatibleVersion() {
@@ -121,7 +123,7 @@ func (s *VersionTestSuite) createKeyspace(keyspace string) func() {
 		NumReplicas:  1,
 		ProtoVersion: environment.GetCassandraProtoVersion(),
 	}
-	client, err := cassandra.NewCQLClient(cfg)
+	client, err := cassandra.NewCQLClient(cfg, gocql.All)
 	s.NoError(err)
 
 	err = client.CreateKeyspace(keyspace)
@@ -167,7 +169,7 @@ func (s *VersionTestSuite) runCheckCompatibleVersion(
 		Password:   environment.GetCassandraPassword(),
 		Keyspace:   keyspace,
 	}
-	err := cassandra.CheckCompatibleVersion(cfg, expected)
+	err := cassandra.CheckCompatibleVersion(cfg, expected, gocql.All)
 	if len(errStr) > 0 {
 		s.Error(err)
 		s.Contains(err.Error(), errStr)

common/persistence/nosql/nosqlplugin/cassandra/tests/utils.go

@@ -21,6 +21,7 @@
 package tests
 
 import (
+	"github.com/uber/cadence/common/persistence/nosql/nosqlplugin/cassandra/gocql"
 	"github.com/uber/cadence/environment"
 	"github.com/uber/cadence/tools/cassandra"
 )
@@ -39,7 +40,7 @@ func NewTestCQLClient(keyspace string) (cassandra.CqlClient, error) {
 		AllowedAuthenticators: environment.GetCassandraAllowedAuthenticators(),
 		NumReplicas:           1,
 		ProtoVersion:          environment.GetCassandraProtoVersion(),
-	})
+	}, gocql.All)
 }
 
 func CreateTestCQLFileContent() string {

tools/cassandra/cqlclient.go

@@ -112,12 +112,16 @@ const (
 var _ schema.SchemaClient = (*CqlClientImpl)(nil)
 
 // NewCQLClient returns a new instance of CQLClient
-func NewCQLClient(cfg *CQLClientConfig) (CqlClient, error) {
+func NewCQLClient(cfg *CQLClientConfig, expectedConsistency gocql.Consistency) (CqlClient, error) {
 	var err error
 
 	cqlClient := new(CqlClientImpl)
 	cqlClient.cfg = cfg
 	cqlClient.nReplicas = cfg.NumReplicas
+	consistency := gocql.All
+	if expectedConsistency == gocql.Quorum {
+		consistency = gocql.Quorum
+	}
 	cqlClient.session, err = gocql.GetRegisteredClient().CreateSession(gocql.ClusterConfig{
 		Hosts:                 cfg.Hosts,
 		Port:                  cfg.Port,
@@ -129,7 +133,7 @@ func NewCQLClient(cfg *CQLClientConfig) (CqlClient, error) {
 		Timeout:               time.Duration(cfg.Timeout) * time.Second,
 		ConnectTimeout:        time.Duration(cfg.ConnectTimeout) * time.Second,
 		ProtoVersion:          cfg.ProtoVersion,
-		Consistency:           gocql.All,
+		Consistency:           consistency,
 	})
 	if err != nil {
 		return nil, err

tools/cassandra/handler.go

@@ -24,6 +24,8 @@ import (
 	"fmt"
 	"log"
 
+	"github.com/uber/cadence/common/persistence/nosql/nosqlplugin/cassandra/gocql"
+
 	"github.com/urfave/cli"
 
 	"github.com/uber/cadence/common/config"
@@ -45,16 +47,17 @@ type SetupSchemaConfig struct {
 // rollback, the code version (expected version) would fall lower than the actual version in
 // cassandra.
 func VerifyCompatibleVersion(
-	cfg config.Persistence,
+	cfg config.Persistence, expectedConsistency gocql.Consistency,
 ) error {
+
 	if ds, ok := cfg.DataStores[cfg.DefaultStore]; ok {
-		if err := verifyCompatibleVersion(ds, cassandra.Version); err != nil {
+		if err := verifyCompatibleVersion(ds, cassandra.Version, expectedConsistency); err != nil {
 			return err
 		}
 	}
 
 	if ds, ok := cfg.DataStores[cfg.VisibilityStore]; ok {
-		if err := verifyCompatibleVersion(ds, cassandra.VisibilityVersion); err != nil {
+		if err := verifyCompatibleVersion(ds, cassandra.VisibilityVersion, expectedConsistency); err != nil {
 			return err
 		}
 	}
@@ -64,14 +67,14 @@ func VerifyCompatibleVersion(
 
 func verifyCompatibleVersion(
 	ds config.DataStore,
-	expectedCassandraVersion string,
+	expectedCassandraVersion string, expectedConsistency gocql.Consistency,
 ) error {
 	if ds.NoSQL != nil {
-		return verifyPluginVersion(ds.NoSQL, expectedCassandraVersion)
+		return verifyPluginVersion(ds.NoSQL, expectedCassandraVersion, expectedConsistency)
 	}
 	if ds.ShardedNoSQL != nil {
 		for shardName, connection := range ds.ShardedNoSQL.Connections {
-			err := verifyPluginVersion(connection.NoSQLPlugin, expectedCassandraVersion)
+			err := verifyPluginVersion(connection.NoSQLPlugin, expectedCassandraVersion, expectedConsistency)
 			if err != nil {
 				return fmt.Errorf("Failed to verify version for DB shard: %v. Error: %v", shardName, err.Error())
 			}
@@ -82,21 +85,22 @@ func verifyCompatibleVersion(
 	return nil
 }
 
-func verifyPluginVersion(plugin *config.NoSQL, expectedCassandraVersion string) error {
+func verifyPluginVersion(plugin *config.NoSQL, expectedCassandraVersion string, expectedConsistency gocql.Consistency) error {
 	// Use hardcoded instead of constant because of cycle dependency issue.
 	// However, this file will be refactor to support NoSQL soon. After the refactoring, cycle dependency issue
 	// should be gone and we can use constant at that time
 	if plugin.PluginName != "cassandra" {
 		return fmt.Errorf("unknown NoSQL plugin name: %q", plugin.PluginName)
 	}
 
-	return CheckCompatibleVersion(*plugin, expectedCassandraVersion)
+	return CheckCompatibleVersion(*plugin, expectedCassandraVersion, expectedConsistency)
 }
 
 // CheckCompatibleVersion check the version compatibility
 func CheckCompatibleVersion(
 	cfg config.Cassandra,
 	expectedVersion string,
+	expectedConsistency gocql.Consistency,
 ) error {
 
 	client, err := NewCQLClient(&CQLClientConfig{
@@ -110,7 +114,7 @@ func CheckCompatibleVersion(
 		ConnectTimeout:        DefaultConnectTimeout,
 		TLS:                   cfg.TLS,
 		ProtoVersion:          cfg.ProtoVersion,
-	})
+	}, expectedConsistency)
 	if err != nil {
 		return fmt.Errorf("creating CQL client: %w", err)
 	}
@@ -127,7 +131,7 @@ func setupSchema(cli *cli.Context) error {
 	if err != nil {
 		return handleErr(schema.NewConfigError(err.Error()))
 	}
-	client, err := NewCQLClient(config)
+	client, err := NewCQLClient(config, gocql.All)
 	if err != nil {
 		return handleErr(err)
 	}
@@ -145,7 +149,7 @@ func updateSchema(cli *cli.Context) error {
 	if err != nil {
 		return handleErr(schema.NewConfigError(err.Error()))
 	}
-	client, err := NewCQLClient(config)
+	client, err := NewCQLClient(config, gocql.All)
 	if err != nil {
 		return handleErr(err)
 	}
@@ -176,7 +180,7 @@ func createKeyspace(cli *cli.Context) error {
 
 func doCreateKeyspace(cfg CQLClientConfig, name string, datacenter string) error {
 	cfg.Keyspace = SystemKeyspace
-	client, err := NewCQLClient(&cfg)
+	client, err := NewCQLClient(&cfg, gocql.All)
 	if err != nil {
 		return err
 	}

tools/cassandra/main.go

@@ -23,6 +23,8 @@ package cassandra
 import (
 	"os"
 
+	"github.com/uber/cadence/common/persistence/nosql/nosqlplugin/cassandra/gocql"
+
 	"github.com/urfave/cli"
 
 	"github.com/uber/cadence/tools/common/schema"
@@ -39,7 +41,7 @@ func SetupSchema(config *SetupSchemaConfig) error {
 	if err := validateCQLClientConfig(&config.CQLClientConfig); err != nil {
 		return err
 	}
-	db, err := NewCQLClient(&config.CQLClientConfig)
+	db, err := NewCQLClient(&config.CQLClientConfig, gocql.All)
 	if err != nil {
 		return err
 	}