Capture CassandraLWT error and log/bump metrics for it.

common/metrics/defs.go

@@ -1810,6 +1810,7 @@ const (
 	PersistenceErrExecutionAlreadyStartedCounter
 	PersistenceErrDomainAlreadyExistsCounter
 	PersistenceErrBadRequestCounter
+	PersistenceErrCassandraLWTCounter
 	PersistenceSampledCounter
 	PersistenceEmptyResponseCounter
 
@@ -2343,6 +2344,7 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
 		PersistenceErrExecutionAlreadyStartedCounter:        {metricName: "persistence_errors_execution_already_started", metricType: Counter},
 		PersistenceErrDomainAlreadyExistsCounter:            {metricName: "persistence_errors_domain_already_exists", metricType: Counter},
 		PersistenceErrBadRequestCounter:                     {metricName: "persistence_errors_bad_request", metricType: Counter},
+		PersistenceErrCassandraLWTCounter:                   {metricName: "persistence_errors_cassandra_lwt", metricType: Counter},
 		PersistenceSampledCounter:                           {metricName: "persistence_sampled", metricType: Counter},
 		PersistenceEmptyResponseCounter:                     {metricName: "persistence_empty_response", metricType: Counter},
 		CadenceClientRequests:                               {metricName: "cadence_client_requests", metricType: Counter},

common/persistence/dataManagerInterfaces.go

@@ -266,6 +266,12 @@ type (
 		Msg string
 	}
 
+	// CassandraLWTError is returned when we get UNAVAILABLE error code from Cassandra with message containing
+	// "Cannot perform LWT operation"
+	CassandraLWTError struct {
+		Msg string
+	}
+
 	// TransactionSizeLimitError is returned when the transaction size is too large
 	TransactionSizeLimitError struct {
 		Msg string
@@ -1862,6 +1868,10 @@ func (e *TimeoutError) Error() string {
 	return e.Msg
 }
 
+func (e *CassandraLWTError) Error() string {
+	return e.Msg
+}
+
 func (e *TransactionSizeLimitError) Error() string {
 	return e.Msg
 }

common/persistence/nosql/nosqlplugin/cassandra/db.go

@@ -68,3 +68,7 @@ func (db *cdb) IsTimeoutError(err error) bool {
 func (db *cdb) IsThrottlingError(err error) bool {
 	return db.client.IsThrottlingError(err)
 }
+
+func (db *cdb) IsCassandraLWTError(err error) bool {
+	return db.client.IsCassandraLWTError(err)
+}

common/persistence/nosql/nosqlplugin/cassandra/gocql/public/client.go

@@ -22,6 +22,7 @@ package public
 
 import (
 	"context"
+	"strings"
 
 	gogocql "github.com/gocql/gocql"
 
@@ -70,3 +71,16 @@ func (c client) IsThrottlingError(err error) bool {
 	}
 	return false
 }
+
+func (c client) IsCassandraLWTError(err error) bool {
+	if req, ok := err.(gogocql.RequestError); ok {
+		// 0x1000 == UNAVAILABLE
+		if req.Code() != 0x1000 {
+			return false
+		}
+		if strings.Contains(req.Message(), "Cannot perform LWT operation") {
+			return true
+		}
+	}
+	return false
+}

common/persistence/nosql/nosqlplugin/interfaces.go

@@ -76,6 +76,7 @@ type (
 		IsTimeoutError(error) bool
 		IsNotFoundError(error) bool
 		IsThrottlingError(error) bool
+		IsCassandraLWTError(error) bool
 	}
 
 	/**

common/persistence/nosql/nosqlplugin/mongodb/error.go

@@ -35,3 +35,7 @@ func (db *mdb) IsTimeoutError(err error) bool {
 func (db *mdb) IsThrottlingError(err error) bool {
 	return false
 }
+
+func (db *mdb) IsCassandraLWTError(err error) bool {
+	return false
+}

common/persistence/nosql/utils.go

@@ -49,6 +49,12 @@ func convertCommonErrors(
 		}
 	}
 
+	if errChecker.IsCassandraLWTError(err) {
+		return &p.CassandraLWTError{
+			Msg: fmt.Sprintf("#{operation} operation failed. Error: #{err}"),
+		}
+	}
+
 	return &types.InternalServiceError{
 		Message: fmt.Sprintf("%v operation failed. Error: %v", operation, err),
 	}

common/persistence/persistenceMetricClients.go

@@ -248,6 +248,10 @@ func (p *persistenceMetricsClientBase) updateErrorMetric(scope int, err error) {
 	case *types.ServiceBusyError:
 		p.metricClient.IncCounter(scope, metrics.PersistenceErrBusyCounter)
 		p.metricClient.IncCounter(scope, metrics.PersistenceFailures)
+	case *CassandraLWTError:
+		p.metricClient.IncCounter(scope, metrics.PersistenceErrCassandraLWTCounter)
+		p.metricClient.IncCounter(scope, metrics.PersistenceFailures)
+		p.logger.Error("Cassandra LWT Error.", tag.Error(err), tag.MetricScope(scope))
 	default:
 		p.logger.Error("Operation failed with internal error.", tag.Error(err), tag.MetricScope(scope))
 		p.metricClient.IncCounter(scope, metrics.PersistenceFailures)