Skip to content

Commit 714aec4

Browse files
agautam478agautam478
authored
[CDNC-4831] Added option to pass consistency level in the cassandra schema versio… (cadence-workflow#5327)
* Added option to pass consistency level in the cassandra schema version validation ---------
1 parent 41c8369 commit 714aec4

File tree

7 files changed

+34
-21
lines changed

7 files changed

+34
-21
lines changed

‎cmd/server/cadence/cadence.go‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,8 @@ import (
2929
"strings"
3030
"syscall"
3131

32+
"github.com/uber/cadence/common/persistence/nosql/nosqlplugin/cassandra/gocql"
33+
3234
"github.com/urfave/cli"
3335

3436
"github.com/uber/cadence/common"
@@ -69,7 +71,7 @@ func startHandler(c *cli.Context) {
6971
log.Fatalf("config validation failed: %v", err)
7072
}
7173
// cassandra schema version validation
72-
if err := cassandra.VerifyCompatibleVersion(cfg.Persistence); err != nil {
74+
if err := cassandra.VerifyCompatibleVersion(cfg.Persistence, gocql.Quorum); err != nil {
7375
log.Fatal("cassandra schema version compatibility check failed: ", err)
7476
}
7577
// sql schema version validation

‎cmd/server/cadence/server_test.go‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,8 @@ import (
2929
"testing"
3030
"time"
3131

32+
"github.com/uber/cadence/common/persistence/nosql/nosqlplugin/cassandra/gocql"
33+
3234
"github.com/uber/cadence/testflags"
3335

3436
"github.com/stretchr/testify/require"
@@ -94,7 +96,7 @@ func (s *ServerSuite) TestServerStartup() {
9496
log.Fatalf("config validation failed: %v", err)
9597
}
9698
// cassandra schema version validation
97-
if err := cassandra.VerifyCompatibleVersion(cfg.Persistence); err != nil {
99+
if err := cassandra.VerifyCompatibleVersion(cfg.Persistence, gocql.All); err != nil {
98100
log.Fatal("cassandra schema version compatibility check failed: ", err)
99101
}
100102

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

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,8 @@ import (
2828
"testing"
2929
"time"
3030

31+
"github.com/uber/cadence/common/persistence/nosql/nosqlplugin/cassandra/gocql"
32+
3133
log "github.com/sirupsen/logrus"
3234
"github.com/stretchr/testify/require"
3335
"github.com/stretchr/testify/suite"
@@ -92,7 +94,7 @@ func (s *VersionTestSuite) TestVerifyCompatibleVersion() {
9294
TransactionSizeLimit: dynamicconfig.GetIntPropertyFn(common.DefaultTransactionSizeLimit),
9395
ErrorInjectionRate: dynamicconfig.GetFloatPropertyFn(0),
9496
}
95-
s.NoError(cassandra.VerifyCompatibleVersion(cfg))
97+
s.NoError(cassandra.VerifyCompatibleVersion(cfg, gocql.All))
9698
}
9799

98100
func (s *VersionTestSuite) TestCheckCompatibleVersion() {
@@ -121,7 +123,7 @@ func (s *VersionTestSuite) createKeyspace(keyspace string) func() {
121123
NumReplicas: 1,
122124
ProtoVersion: environment.GetCassandraProtoVersion(),
123125
}
124-
client, err := cassandra.NewCQLClient(cfg)
126+
client, err := cassandra.NewCQLClient(cfg, gocql.All)
125127
s.NoError(err)
126128

127129
err = client.CreateKeyspace(keyspace)
@@ -167,7 +169,7 @@ func (s *VersionTestSuite) runCheckCompatibleVersion(
167169
Password: environment.GetCassandraPassword(),
168170
Keyspace: keyspace,
169171
}
170-
err := cassandra.CheckCompatibleVersion(cfg, expected)
172+
err := cassandra.CheckCompatibleVersion(cfg, expected, gocql.All)
171173
if len(errStr) > 0 {
172174
s.Error(err)
173175
s.Contains(err.Error(), errStr)

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
package tests
2222

2323
import (
24+
"github.com/uber/cadence/common/persistence/nosql/nosqlplugin/cassandra/gocql"
2425
"github.com/uber/cadence/environment"
2526
"github.com/uber/cadence/tools/cassandra"
2627
)
@@ -39,7 +40,7 @@ func NewTestCQLClient(keyspace string) (cassandra.CqlClient, error) {
3940
AllowedAuthenticators: environment.GetCassandraAllowedAuthenticators(),
4041
NumReplicas: 1,
4142
ProtoVersion: environment.GetCassandraProtoVersion(),
42-
})
43+
}, gocql.All)
4344
}
4445

4546
func CreateTestCQLFileContent() string {

‎tools/cassandra/cqlclient.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -112,7 +112,7 @@ const (
112112
var _ schema.SchemaClient = (*CqlClientImpl)(nil)
113113

114114
// NewCQLClient returns a new instance of CQLClient
115-
func NewCQLClient(cfg *CQLClientConfig) (CqlClient, error) {
115+
func NewCQLClient(cfg *CQLClientConfig, expectedConsistency gocql.Consistency) (CqlClient, error) {
116116
var err error
117117

118118
cqlClient := new(CqlClientImpl)
@@ -129,7 +129,7 @@ func NewCQLClient(cfg *CQLClientConfig) (CqlClient, error) {
129129
Timeout: time.Duration(cfg.Timeout) * time.Second,
130130
ConnectTimeout: time.Duration(cfg.ConnectTimeout) * time.Second,
131131
ProtoVersion: cfg.ProtoVersion,
132-
Consistency: gocql.All,
132+
Consistency: expectedConsistency,
133133
})
134134
if err != nil {
135135
return nil, err

‎tools/cassandra/handler.go‎

Lines changed: 16 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@ import (
2424
"fmt"
2525
"log"
2626

27+
"github.com/uber/cadence/common/persistence/nosql/nosqlplugin/cassandra/gocql"
28+
2729
"github.com/urfave/cli"
2830

2931
"github.com/uber/cadence/common/config"
@@ -45,16 +47,17 @@ type SetupSchemaConfig struct {
4547
// rollback, the code version (expected version) would fall lower than the actual version in
4648
// cassandra.
4749
func VerifyCompatibleVersion(
48-
cfg config.Persistence,
50+
cfg config.Persistence, expectedConsistency gocql.Consistency,
4951
) error {
52+
5053
if ds, ok := cfg.DataStores[cfg.DefaultStore]; ok {
51-
if err := verifyCompatibleVersion(ds, cassandra.Version); err != nil {
54+
if err := verifyCompatibleVersion(ds, cassandra.Version, expectedConsistency); err != nil {
5255
return err
5356
}
5457
}
5558

5659
if ds, ok := cfg.DataStores[cfg.VisibilityStore]; ok {
57-
if err := verifyCompatibleVersion(ds, cassandra.VisibilityVersion); err != nil {
60+
if err := verifyCompatibleVersion(ds, cassandra.VisibilityVersion, expectedConsistency); err != nil {
5861
return err
5962
}
6063
}
@@ -64,14 +67,14 @@ func VerifyCompatibleVersion(
6467

6568
func verifyCompatibleVersion(
6669
ds config.DataStore,
67-
expectedCassandraVersion string,
70+
expectedCassandraVersion string, expectedConsistency gocql.Consistency,
6871
) error {
6972
if ds.NoSQL != nil {
70-
return verifyPluginVersion(ds.NoSQL, expectedCassandraVersion)
73+
return verifyPluginVersion(ds.NoSQL, expectedCassandraVersion, expectedConsistency)
7174
}
7275
if ds.ShardedNoSQL != nil {
7376
for shardName, connection := range ds.ShardedNoSQL.Connections {
74-
err := verifyPluginVersion(connection.NoSQLPlugin, expectedCassandraVersion)
77+
err := verifyPluginVersion(connection.NoSQLPlugin, expectedCassandraVersion, expectedConsistency)
7578
if err != nil {
7679
return fmt.Errorf("Failed to verify version for DB shard: %v. Error: %v", shardName, err.Error())
7780
}
@@ -82,21 +85,22 @@ func verifyCompatibleVersion(
8285
return nil
8386
}
8487

85-
func verifyPluginVersion(plugin *config.NoSQL, expectedCassandraVersion string) error {
88+
func verifyPluginVersion(plugin *config.NoSQL, expectedCassandraVersion string, expectedConsistency gocql.Consistency) error {
8689
// Use hardcoded instead of constant because of cycle dependency issue.
8790
// However, this file will be refactor to support NoSQL soon. After the refactoring, cycle dependency issue
8891
// should be gone and we can use constant at that time
8992
if plugin.PluginName != "cassandra" {
9093
return fmt.Errorf("unknown NoSQL plugin name: %q", plugin.PluginName)
9194
}
9295

93-
return CheckCompatibleVersion(*plugin, expectedCassandraVersion)
96+
return CheckCompatibleVersion(*plugin, expectedCassandraVersion, expectedConsistency)
9497
}
9598

9699
// CheckCompatibleVersion check the version compatibility
97100
func CheckCompatibleVersion(
98101
cfg config.Cassandra,
99102
expectedVersion string,
103+
expectedConsistency gocql.Consistency,
100104
) error {
101105

102106
client, err := NewCQLClient(&CQLClientConfig{
@@ -110,7 +114,7 @@ func CheckCompatibleVersion(
110114
ConnectTimeout: DefaultConnectTimeout,
111115
TLS: cfg.TLS,
112116
ProtoVersion: cfg.ProtoVersion,
113-
})
117+
}, expectedConsistency)
114118
if err != nil {
115119
return fmt.Errorf("creating CQL client: %w", err)
116120
}
@@ -127,7 +131,7 @@ func setupSchema(cli *cli.Context) error {
127131
if err != nil {
128132
return handleErr(schema.NewConfigError(err.Error()))
129133
}
130-
client, err := NewCQLClient(config)
134+
client, err := NewCQLClient(config, gocql.All)
131135
if err != nil {
132136
return handleErr(err)
133137
}
@@ -145,7 +149,7 @@ func updateSchema(cli *cli.Context) error {
145149
if err != nil {
146150
return handleErr(schema.NewConfigError(err.Error()))
147151
}
148-
client, err := NewCQLClient(config)
152+
client, err := NewCQLClient(config, gocql.All)
149153
if err != nil {
150154
return handleErr(err)
151155
}
@@ -176,7 +180,7 @@ func createKeyspace(cli *cli.Context) error {
176180

177181
func doCreateKeyspace(cfg CQLClientConfig, name string, datacenter string) error {
178182
cfg.Keyspace = SystemKeyspace
179-
client, err := NewCQLClient(&cfg)
183+
client, err := NewCQLClient(&cfg, gocql.All)
180184
if err != nil {
181185
return err
182186
}

‎tools/cassandra/main.go‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@ package cassandra
2323
import (
2424
"os"
2525

26+
"github.com/uber/cadence/common/persistence/nosql/nosqlplugin/cassandra/gocql"
27+
2628
"github.com/urfave/cli"
2729

2830
"github.com/uber/cadence/tools/common/schema"
@@ -39,7 +41,7 @@ func SetupSchema(config *SetupSchemaConfig) error {
3941
if err := validateCQLClientConfig(&config.CQLClientConfig); err != nil {
4042
return err
4143
}
42-
db, err := NewCQLClient(&config.CQLClientConfig)
44+
db, err := NewCQLClient(&config.CQLClientConfig, gocql.All)
4345
if err != nil {
4446
return err
4547
}

0 commit comments

Comments
 (0)