tunable_retry_policy.py

class TunableRetryPolicy(RetryPolicy):
    """ A retry policy that allows you to decide if you want to downgrade consistency before
    attempting to retry, additionally, you can specify the number of retries to try
    """

    def __init__(self, read_retries=3, write_retries=3, downgrade_consistency=True):
        self._read_retries = read_retries
        self._write_retries = write_retries
        self._downgrade_consistency = downgrade_consistency

    def _pick_consistency(self, num_responses):
        if num_responses >= 3:
            return self.RETRY, ConsistencyLevel.THREE
        elif num_responses >= 2:
            return self.RETRY, ConsistencyLevel.TWO
        else:
            return self.RETRY, ConsistencyLevel.ONE

    def on_read_timeout(self, query, consistency, required_responses,
                        received_responses, data_retrieved, retry_num):
        """
        This is called when the read times out, this will retry _read_retries times downgrading
        the consistency level (if allowed) until the read succeeds or until _read_retries is
        exhausted.
        :param query:
        :param consistency:
        :param required_responses:
        :param received_responses:
        :param data_retrieved:
        :param retry_num:
        :return:
        """
        logging.warn('Read Retry {0} of {1}'.format(retry_num, self._read_retries))
        if retry_num >= self._read_retries:
            return self.RETHROW, None
        elif received_responses < required_responses and self._downgrade_consistency:
            return self._pick_consistency(received_responses)
        elif not data_retrieved:
            return self.RETRY, consistency
        else:
            return self.RETHROW, None

    def on_write_timeout(self, query, consistency, write_type,
                         required_responses, received_responses, retry_num):
        """
        This is called when the write times out, this will retry _write_retries times downgrading
        the consistency level (if allowed) until the write succeeds or until _write_retries is
        exhausted. If the write type is not idempotent, we will ignore it.
        :param query:
        :param consistency:
        :param write_type:
        :param required_responses:
        :param received_responses:
        :param retry_num:
        :return:
        """
        logging.warn('Write Retry {0} of {1}'.format(retry_num, self._read_retries))
        if retry_num >= self._write_retries:
            return self.RETHROW, None
        elif write_type == WriteType.COUNTER:
            return self.IGNORE, None
        elif self._downgrade_consistency:
            return self._pick_consistency(received_responses)
        else:
            return self.RETHROW, None

    def on_unavailable(self, query, consistency, required_replicas, alive_replicas, retry_num):
        """Called when we get an Unavailable Error that means one or more replicas are down.
        There's no point in retrying this more than once if we decide to allow downgrading of
        the consistency level.
        """
        logging.warn('Unavailable Exception required:{0} alive:{1}'.format(required_replicas,
                                                                           alive_replicas))
        if retry_num >= 0:
            return self.RETHROW, None
        elif self._downgrade_consistency:
            return self._pick_consistency(alive_replicas)
        else:
            return self.RETHROW, None