Integration tests: Add DDL retry

In the previous PR I forgot to also modify the utils in integration tests.

Author: Lorak-mmk

scylla/tests/integration/utils.rs

@@ -5,8 +5,9 @@ use scylla::client::session_builder::{GenericSessionBuilder, SessionBuilderKind}
 use scylla::cluster::ClusterState;
 use scylla::cluster::NodeRef;
 use scylla::deserialize::DeserializeValue;
-use scylla::errors::ExecutionError;
+use scylla::errors::{DbError, ExecutionError, RequestAttemptError};
 use scylla::policies::load_balancing::{FallbackPlan, LoadBalancingPolicy, RoutingInfo};
+use scylla::policies::retry::{RequestInfo, RetryDecision, RetryPolicy, RetrySession};
 use scylla::query::Query;
 use scylla::routing::Shard;
 use std::collections::HashMap;
@@ -17,6 +18,7 @@ use std::str::FromStr;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
+use tracing::{error, warn};
 
 use scylla_proxy::{Node, Proxy, ProxyError, RunningProxy, ShardAwareness};
 
@@ -207,12 +209,54 @@ impl LoadBalancingPolicy for SchemaQueriesLBP {
     }
 }
 
+#[derive(Debug, Default)]
+struct SchemaQueriesRetrySession {
+    count: usize,
+}
+
+impl RetrySession for SchemaQueriesRetrySession {
+    fn decide_should_retry(&mut self, request_info: RequestInfo) -> RetryDecision {
+        match request_info.error {
+            RequestAttemptError::DbError(DbError::ServerError, s)
+                if s == "Failed to apply group 0 change due to concurrent modification" =>
+            {
+                self.count += 1;
+                // Give up if there are many failures.
+                // In this case we really should do something about it in the
+                // core, because it is absurd for DDL queries to fail this often.
+                if self.count >= 10 {
+                    error!("Received TENTH(!) group 0 concurrent modification error during DDL. Please fix Scylla Core.");
+                    RetryDecision::DontRetry
+                } else {
+                    warn!("Received group 0 concurrent modification error during DDL. Performing retry #{}.", self.count);
+                    RetryDecision::RetrySameNode(None)
+                }
+            }
+            _ => RetryDecision::DontRetry,
+        }
+    }
+
+    fn reset(&mut self) {
+        *self = Default::default()
+    }
+}
+
+#[derive(Debug)]
+struct SchemaQueriesRetryPolicy;
+
+impl RetryPolicy for SchemaQueriesRetryPolicy {
+    fn new_session(&self) -> Box<dyn RetrySession> {
+        Box::new(SchemaQueriesRetrySession::default())
+    }
+}
+
 fn apply_ddl_lbp(query: &mut Query) {
     let policy = query
         .get_execution_profile_handle()
         .map(|profile| profile.pointee_to_builder())
         .unwrap_or(ExecutionProfile::builder())
         .load_balancing_policy(Arc::new(SchemaQueriesLBP))
+        .retry_policy(Arc::new(SchemaQueriesRetryPolicy))
         .build();
     query.set_execution_profile_handle(Some(policy.into_handle()));
 }