diff --git a/cassandra/cluster.py b/cassandra/cluster.py
index 66bf7c7049..99f806fd66 100644
--- a/cassandra/cluster.py
+++ b/cassandra/cluster.py
@@ -4780,6 +4780,10 @@ def _set_result(self, host, connection, pool, response):
                     log.warning("Host %s error: %s.", host, response.summary)
                     if self._metrics is not None:
                         self._metrics.on_other_error()
+                    # TruncateError should not be retried as it indicates a permanent failure
+                    if isinstance(response, TruncateError):
+                        self._set_final_exception(response.to_exception())
+                        return
                     cl = getattr(self.message, 'consistency_level', None)
                     retry = retry_policy.on_request_error(
                         self.query, cl, error=response,
diff --git a/tests/integration/simulacron/test_policies.py b/tests/integration/simulacron/test_policies.py
index 3f94a41222..9e16caa646 100644
--- a/tests/integration/simulacron/test_policies.py
+++ b/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  # 3 calls (excludes TruncateError)
 
-        # 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
diff --git a/tests/unit/test_truncate_error.py b/tests/unit/test_truncate_error.py
new file mode 100644
index 0000000000..223fe0770e
--- /dev/null
+++ b/tests/unit/test_truncate_error.py
@@ -0,0 +1,59 @@
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+from cassandra.protocol import TruncateError
+
+
+class TruncateErrorTest(unittest.TestCase):
+    """
+    Test TruncateError exception behavior
+    """
+
+    def test_truncate_error_message_includes_server_message(self):
+        """
+        Verify that TruncateError includes the server-provided error message
+        in addition to the generic summary.
+        
+        This addresses the issue where the error message was thought to be ignored.
+        The server message should always be included in the error string representation.
+        """
+        server_message = "unconfigured table test_table"
+        error = TruncateError(code=0x1003, message=server_message, info=None)
+        error_str = str(error)
+        
+        # Verify both the summary and server message are in the error string
+        self.assertIn("Error during truncate", error_str, 
+                     "Generic summary should be in error string")
+        self.assertIn(server_message, error_str,
+                     "Server-provided message should be in error string")
+        
+    def test_truncate_error_code(self):
+        """
+        Verify that TruncateError has the correct error code (0x1003)
+        """
+        error = TruncateError(code=0x1003, message="test", info=None)
+        self.assertEqual(error.code, 0x1003)
+        self.assertEqual(error.error_code, 0x1003)
+
+    def test_truncate_error_summary(self):
+        """
+        Verify that TruncateError has the correct summary message
+        """
+        error = TruncateError(code=0x1003, message="test", info=None)
+        self.assertEqual(error.summary, "Error during truncate")
+
+
+if __name__ == '__main__':
+    unittest.main()