Fix domain replication queue cleanup query (#3259)

uber · May 13, 2020 · 25bb63e · 25bb63e
1 parent 9821fb0
commit 25bb63e
Showing 1 changed file with 11 additions and 11 deletions.
diff --git a/common/persistence/cassandra/cassandraQueue.go b/common/persistence/cassandra/cassandraQueue.go
@@ -39,15 +39,16 @@ const (
 )
 
 const (
-	templateEnqueueMessageQuery      = `INSERT INTO queue (queue_type, message_id, message_payload) VALUES(?, ?, ?) IF NOT EXISTS`
-	templateGetLastMessageIDQuery    = `SELECT message_id FROM queue WHERE queue_type=? ORDER BY message_id DESC LIMIT 1`
-	templateGetMessagesQuery         = `SELECT message_id, message_payload FROM queue WHERE queue_type = ? and message_id > ? LIMIT ?`
-	templateGetMessagesFromDLQQuery  = `SELECT message_id, message_payload FROM queue WHERE queue_type = ? and message_id > ? and message_id <= ?`
-	templateDeleteMessagesQuery      = `DELETE FROM queue WHERE queue_type = ? and message_id > ? and message_id <= ?`
-	templateDeleteMessageQuery       = `DELETE FROM queue WHERE queue_type = ? and message_id = ?`
-	templateGetQueueMetadataQuery    = `SELECT cluster_ack_level, version FROM queue_metadata WHERE queue_type = ?`
-	templateInsertQueueMetadataQuery = `INSERT INTO queue_metadata (queue_type, cluster_ack_level, version) VALUES(?, ?, ?) IF NOT EXISTS`
-	templateUpdateQueueMetadataQuery = `UPDATE queue_metadata SET cluster_ack_level = ?, version = ? WHERE queue_type = ? IF version = ?`
+	templateEnqueueMessageQuery       = `INSERT INTO queue (queue_type, message_id, message_payload) VALUES(?, ?, ?) IF NOT EXISTS`
+	templateGetLastMessageIDQuery     = `SELECT message_id FROM queue WHERE queue_type=? ORDER BY message_id DESC LIMIT 1`
+	templateGetMessagesQuery          = `SELECT message_id, message_payload FROM queue WHERE queue_type = ? and message_id > ? LIMIT ?`
+	templateGetMessagesFromDLQQuery   = `SELECT message_id, message_payload FROM queue WHERE queue_type = ? and message_id > ? and message_id <= ?`
+	templateDeleteMessagesBeforeQuery = `DELETE FROM queue WHERE queue_type = ? and message_id < ?`
+	templateDeleteMessagesQuery       = `DELETE FROM queue WHERE queue_type = ? and message_id > ? and message_id <= ?`
+	templateDeleteMessageQuery        = `DELETE FROM queue WHERE queue_type = ? and message_id = ?`
+	templateGetQueueMetadataQuery     = `SELECT cluster_ack_level, version FROM queue_metadata WHERE queue_type = ?`
+	templateInsertQueueMetadataQuery  = `INSERT INTO queue_metadata (queue_type, cluster_ack_level, version) VALUES(?, ?, ?) IF NOT EXISTS`
+	templateUpdateQueueMetadataQuery  = `UPDATE queue_metadata SET cluster_ack_level = ?, version = ? WHERE queue_type = ? IF version = ?`
 )
 
 type (
@@ -295,13 +296,12 @@ func (q *cassandraQueue) DeleteMessagesBefore(
 	messageID int64,
 ) error {
 
-	query := q.session.Query(templateDeleteMessagesQuery, q.queueType, messageID)
+	query := q.session.Query(templateDeleteMessagesBeforeQuery, q.queueType, messageID)
 	if err := query.Exec(); err != nil {
 		return &workflow.InternalServiceError{
 			Message: fmt.Sprintf("DeleteMessagesBefore operation failed. Error %v", err),
 		}
 	}
-
 	return nil
 }