Fix TruncateError retry behavior - stop retrying on truncate errors

Co-authored-by: mykaul <4655593+mykaul@users.noreply.github.com>

Author: Copilot

cassandra/cluster.py

@@ -5257,14 +5257,24 @@ def _set_result(self, host, connection, pool, response):
                         self.query, retry_num=self._query_retries, **response.info)
                 elif isinstance(response, (OverloadedErrorMessage,
                                            IsBootstrappingErrorMessage,
-                                           TruncateError, ServerError)):
+                                           ServerError)):
                     log.warning("Host %s error: %s.", host, response.summary)
                     if self._metrics is not None:
                         self._metrics.on_other_error()
                     cl = getattr(self.message, 'consistency_level', None)
                     retry = retry_policy.on_request_error(
                         self.query, cl, error=response,
                         retry_num=self._query_retries)
+                elif isinstance(response, TruncateError):
+                    # TruncateError should not be retried as it indicates a permanent failure
+                    log.warning("Host %s truncate error: %s.", host, response.summary)
+                    if self._metrics is not None:
+                        self._metrics.on_other_error()
+                    if hasattr(response, 'to_exception'):
+                        self._set_final_exception(response.to_exception())
+                    else:
+                        self._set_final_exception(response)
+                    return
                 elif isinstance(response, PreparedQueryNotFound):
                     if self.prepared_statement:
                         query_id = self.prepared_statement.query_id

tests/integration/simulacron/test_policies.py

@@ -424,14 +424,13 @@ def test_retry_policy_on_request_error(self):
             "message": "server_error"
         }
 
-        # Test the on_request_error call
+        # Test the on_request_error call for errors that should retry
         retry_policy = CounterRetryPolicy()
         self.set_cluster(retry_policy)
 
         for prime_error, exc in [
             (overloaded_error, OverloadedErrorMessage),
             (bootstrapping_error, IsBootstrappingErrorMessage),
-            (truncate_error, TruncateError),
             (server_error, ServerError)]:
 
             clear_queries()
@@ -444,13 +443,24 @@ def test_retry_policy_on_request_error(self):
 
             assert len(rf.attempted_hosts) == 1  # no retry
 
-        assert next(retry_policy.request_error) == 4
+        assert next(retry_policy.request_error) == 3
 
-        # Test that by default, retry on next host
+        # Test TruncateError is not retried (no on_request_error call)
+        clear_queries()
+        query_to_prime = "SELECT * from simulacron_keyspace.simulacron_table;"
+        prime_query(query_to_prime, then=truncate_error, rows=None, column_types=None)
+        rf = self.session.execute_async(query_to_prime)
+
+        with pytest.raises(TruncateError):
+            rf.result()
+
+        assert len(rf.attempted_hosts) == 1  # no retry
+
+        # Test that by default, errors retry on next host (except TruncateError)
         retry_policy = RetryPolicy()
         self.set_cluster(retry_policy)
 
-        for e in [overloaded_error, bootstrapping_error, truncate_error, server_error]:
+        for e in [overloaded_error, bootstrapping_error, server_error]:
             clear_queries()
             query_to_prime = "SELECT * from simulacron_keyspace.simulacron_table;"
             prime_query(query_to_prime, then=e, rows=None, column_types=None)
@@ -460,3 +470,14 @@ def test_retry_policy_on_request_error(self):
                 rf.result()
 
             assert len(rf.attempted_hosts) == 3  # all 3 nodes failed
+
+        # Test TruncateError does not retry even with default RetryPolicy
+        clear_queries()
+        query_to_prime = "SELECT * from simulacron_keyspace.simulacron_table;"
+        prime_query(query_to_prime, then=truncate_error, rows=None, column_types=None)
+        rf = self.session.execute_async(query_to_prime)
+
+        with pytest.raises(TruncateError):
+            rf.result()
+
+        assert len(rf.attempted_hosts) == 1  # no retry for TruncateError