Don't use idempotent retry policy on LWT delete
LWT statements need to be retried manually.
@@ -4,7 +4,10 @@ import re
import sys
from cassandra.concurrent import execute_concurrent
-from appscale.datastore.cassandra_env.retry_policies import BASIC_RETRIES
+from appscale.datastore.cassandra_env.retry_policies import (
+ BASIC_RETRIES,
+ NO_RETRIES
+)
from appscale.datastore.dbconstants import TRANSIENT_CASSANDRA_ERRORS
from cassandra.query import BatchStatement
from cassandra.query import ConsistencyLevel
@@ -940,7 +943,7 @@ class PullQueue(Queue):
'id': task_id}
self.db_access.session.execute(delete_index, parameters)
- def _delete_task_and_index(self, task):
+ def _delete_task_and_index(self, task, retries=5):
""" Deletes a task and its index atomically.
Args:
@@ -950,9 +953,15 @@ class PullQueue(Queue):
DELETE FROM pull_queue_tasks
WHERE app = %(app)s AND queue = %(queue)s AND id = %(id)s
IF EXISTS
- """)
+ """, retry_policy=NO_RETRIES)
parameters = {'app': self.app, 'queue': self.name, 'id': task.id}
+ try:
self.db_access.session.execute(delete_task, parameters=parameters)
+ except TRANSIENT_CASSANDRA_ERRORS:
+ retries_left = retries - 1
+ if retries_left <= 0:
+ raise
+ return self._delete_task_and_index(task, retries=retries_left)
delete_task_index = SimpleStatement("""
DELETE FROM pull_queue_tasks_index
