cdrs-tokio: High CPU usage for cdrs_tokio::transport::AsyncTransport::start_processing

I have setup a cassandra cluster of 6 pods using the helm charts and I am connecting to that cluster with one known node. The application works fine, but occasionally the cpu usage of my application would go full and the application becomes unresponsive. I sampled the cpu usage at one of the peak cpu usage for 1 second and found that 99% of cpu is being used in cdrs_tokio::transport::AsyncTransport::start_processing, I have attached the complete flamegraph SVG for your reference here. I also observe that after few hours of full cpu usage, the usage does comes down to 50% (but still high)

The flamegraph was captured without debug symbols since It happens occasionally, below is the code that connect demonstrate how I make the connection, here CASSANDRA_HOST is the name of the service of K8 application.

pub async fn create_cassandra_session() -> anyhow::Result<CassandraSession> {
    let url = format!(
        "{}:{}",
        std::env::var("CASSANDRA_HOST").unwrap_or("127.0.0.1".to_string()),
        std::env::var("CASSANDRA_PORT").unwrap_or("9042".to_string())
    );
    tracing::info!("Connecting to cassandra at {url}");
    let cluster_config = if let (Some(username), Some(password)) = (
        std::env::var("CASSANDRA_USERNAME").ok(),
        std::env::var("CASSANDRA_PASSWORD").ok(),
    ) {
        tracing::info!("Cassandra static credentials read from environment");
        let authenticator = StaticPasswordAuthenticatorProvider::new(username, password);
        NodeTcpConfigBuilder::new()
            .with_contact_point(url.into())
            .with_authenticator_provider(Arc::new(authenticator))
            .build()
            .await?
    } else {
        tracing::info!("Cassandra is connecting without credentials");
        NodeTcpConfigBuilder::new()
            .with_contact_point(url.into())
            .build()
            .await?
    };

    Ok(TcpSessionBuilder::new(RoundRobinLoadBalancingStrategy::new(), cluster_config).build()?)
}