Fix domain replication queue cleanup query
@@ -43,6 +43,7 @@ const (
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 = ?`
@@ -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
}
