Reconnecting node to cluster breaks gocql.Session

[isopov]

Sorry for the lengthy reproduction (maybe it is possible to make it smaller and have more control)
What I do:

• Install bitnami/cassandra to my local k8s (I use kind) helm install --namespace cassandra --create-namespace cassandra bitnami/cassandra --set "replicaCount=6"
• Expose it to connect from localhost kubectl expose service cassandra --type=NodePort --name=cassandra-external --namespace=cassandra
• Get host for connection with docker inspect kind-control-plane | grep IPAddress
• Get port for connection with kubectl get services --namespace=cassandra | grep external
• Get password with kubectl get secret --namespace "cassandra" cassandra -o jsonpath="{.data.cassandra-password}" | base64 -d
• Run this code:

package main

import (
	"fmt"
	"math/rand/v2"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	"github.com/gocql/gocql"
)

const workers = 100
const queries = 10_000

const readWorkers = 100
const readQueries = 10_000_000

// helm install --namespace cassandra --create-namespace cassandra bitnami/cassandra --set "replicaCount=6"
// kubectl expose service cassandra --type=NodePort --name=cassandra-external --namespace=cassandra
func main() {
	//host from docker inspect kind-control-plane | grep IPAddress
	//port from kubectl get services --namespace=cassandra | grep external
	cluster := gocql.NewCluster("172.18.0.2:32146")
	cluster.Authenticator = gocql.PasswordAuthenticator{
		Username: "cassandra",
		//from kubectl get secret --namespace "cassandra" cassandra -o jsonpath="{.data.cassandra-password}" | base64 -d
		Password: "kMDfXIiu5M",
	}
	session, err := cluster.CreateSession()
	if err != nil {
		panic(err)
	}

	execRelease(session.Query("drop keyspace if exists k8stest"))
	execRelease(session.Query("create keyspace k8stest with replication = {'class' : 'SimpleStrategy', 'replication_factor' : 3}"))
	execRelease(session.Query("drop table if exists k8stest.test"))
	execRelease(session.Query("create table k8stest.test (a int, b text, primary key(a))"))

	var wg sync.WaitGroup

	for i := 0; i <= workers; i++ {
		wg.Add(1)

		go func() {
			defer wg.Done()
			query := session.Query("insert into k8stest.test (a, b) values (?,?)")
			for j := i * queries; j < (i+1)*queries; j++ {
				query.Bind(j, "Message"+strconv.Itoa(j))
				if err := query.Exec(); err != nil {
					panic(err)
				}
			}
			query.Release()
		}()
	}

	wg.Wait()

	var scans uint64
	var errors uint64

	var mu sync.RWMutex
	erorsCount := make(map[string]uint64)

	for i := 0; i <= readWorkers; i++ {
		wg.Add(1)

		go func() {
			defer wg.Done()
			query := session.Query("select b from k8stest.test where a=?")
			for j := i * readQueries; j < (i+1)*readQueries; j++ {
				id := rand.IntN(queries * workers)
				query.Bind(id)
				iter := query.Iter()
				var val string
				if iter.Scan(&val) {
					if val != "Message"+strconv.Itoa(id) {
						panic("unexpected message " + val + "instead of Message" + strconv.Itoa(id))
					}
					atomic.AddUint64(&scans, 1)
				} else {
					atomic.AddUint64(&errors, 1)
					time.Sleep(time.Millisecond)
				}
				if err := iter.Close(); err != nil {
					mu.Lock()
					erorsCount[err.Error()]++
					mu.Unlock()
					query.Release()
					query = session.Query("select b from k8stest.test where a=?")
				}

			}
			query.Release()
		}()
	}

	go func() {
		for {
			time.Sleep(time.Second)
			fmt.Printf("##### %d scans, %d errors\n", scans, errors)
			mu.RLock()
			for err, count := range erorsCount {
				fmt.Printf("error %s count %d \n", err, count)
			}
			mu.RUnlock()
		}
	}()

	wg.Wait()
}

func execRelease(query *gocql.Query) {
	if err := query.Exec(); err != nil {
		println(err.Error())
		panic(err)
	}
	query.Release()
}

• After reading starts:

##### 4494 scans, 0 errors
##### 11271 scans, 0 errors
##### 17814 scans, 0 errors
##### 23093 scans, 0 errors
...

I remove one pod from the statefulset

• Get a bit of errors and reading continues

##### 373219 scans, 685 errors
error Server is shutting down count 635 
error writev tcp 172.18.0.1:57908->172.18.0.2:32146: writev: broken pipe count 50

• However after cassandra node joins the ring session breaks and I receive only errors:

##### 546274 scans, 685 errors
error Server is shutting down count 635 
error writev tcp 172.18.0.1:57908->172.18.0.2:32146: writev: broken pipe count 50 
##### 553261 scans, 685 errors
error Server is shutting down count 635 
error writev tcp 172.18.0.1:57908->172.18.0.2:32146: writev: broken pipe count 50 
##### 556674 scans, 41433 errors
error gocql: connection closed waiting for response count 101 
error gocql: no hosts available in the pool count 40547 
error Server is shutting down count 635 
error writev tcp 172.18.0.1:57908->172.18.0.2:32146: writev: broken pipe count 50 
##### 556674 scans, 121771 errors
error gocql: connection closed waiting for response count 101 
error gocql: no hosts available in the pool count 120903 
error Server is shutting down count 635 
error writev tcp 172.18.0.1:57908->172.18.0.2:32146: writev: broken pipe count 50 
...
##### 556674 scans, 10370862 errors
error Server is shutting down count 635 
error writev tcp 172.18.0.1:57908->172.18.0.2:32146: writev: broken pipe count 50 
error gocql: connection closed waiting for response count 101 
error gocql: no hosts available in the pool count 10369976

I expect gocql.Session to continue serving queries. It seems that this reaction is equal for removing and waiting for recreation of any node in cluster.

[isopov]

Similar test with java driver:

package io.github.isopov.cassandra;

import com.datastax.oss.driver.api.core.CqlSession;

import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.LongAdder;
import java.util.random.RandomGenerator;

public class Main {

    public static final int WORKERS = 100;
    public static final int QUERIES = 10_000;


    public static final int READ_WORKERS = 100;
    //    public static final int READ_QUERIES = 100;
    public static final int READ_QUERIES = 10_000_000;

    public static void main(String[] args) {
        try (var session = CqlSession.builder()
                .addContactPoint(new InetSocketAddress("172.18.0.2", 32146))
                //from kubectl get secret --namespace "cassandra" cassandra -o jsonpath="{.data.cassandra-password}" | base64 -d
                .withAuthCredentials("cassandra", "kMDfXIiu5M")
                .withLocalDatacenter("datacenter1")
                .build()) {

            session.execute("drop keyspace if exists k8stest2");
            session.execute("create keyspace k8stest2 with replication = {'class' : 'SimpleStrategy', 'replication_factor' : 3}");
            session.execute("use k8stest2");
            session.execute("drop table if exists test");
            session.execute("create table test (a int, b text, primary key(a))");

            try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
                for (int i = 0; i < WORKERS; i++) {
                    final var fi = i;
                    executor.submit(() -> {
                        try {
                            var query = session.prepare("insert into test (a, b) values (?,?)");
                            for (int j = fi * QUERIES; j < (fi + 1) * QUERIES; j++) {
                                session.execute(query.bind(j, "Message" + j));
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                            throw e;
                        }
                    });
                }
            }

            var reads = new LongAdder();
            var errors = new LongAdder();
            var errorsCount = new ConcurrentHashMap<String, Long>();

            try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
                for (int i = 0; i < READ_WORKERS; i++) {
                    final var fi = i;
                    executor.submit(() -> {
                        RandomGenerator random = RandomGenerator.getDefault();
                        var query = session.prepare("select b from test where a=?");
                        for (int j = fi * READ_QUERIES; j < (fi + 1) * READ_QUERIES; j++) {
                            try {
                                var id = random.nextInt(QUERIES * WORKERS);
                                var res = session.execute(query.bind(id)).all();
                                if (res.size() != 1) {
                                    throw new IllegalStateException("not read without exception");
                                }
                                reads.increment();

                            } catch (Exception e) {
                                errors.increment();
                                errorsCount.compute(e.getMessage(), (msg, count) -> {
                                    if (count == null) {
                                        return 1L;
                                    }
                                    return count + 1L;
                                });
                                try {
                                    Thread.sleep(Duration.ofMillis(1));
                                } catch (InterruptedException ex) {
                                    throw new RuntimeException(ex);
                                }
                            }
                        }
                    });
                }
                executor.submit(() -> {
                    while (true) {
                        Thread.sleep(Duration.ofSeconds(1));
                        System.out.println("##### " + reads + " scans, " + errors + " errors");
                        errorsCount.forEach((msg, count) -> {
                            System.out.println("error " + msg + " count " + count);
                        });
                    }
                });
            }
        }
        System.out.println("All is done!");
    }
}

Output with removing of host during read phase

##### 1784 scans, 0 errors
##### 4260 scans, 0 errors
##### 6605 scans, 0 errors
##### 9168 scans, 0 errors
##### 11809 scans, 0 errors
...
##### 525279 scans, 33550 errors
error Node(endPoint=/172.18.0.2:32146, hostId=35a8ed41-4a70-4945-9eec-53a299c03fb7, hashCode=683c0c99) is overloaded: Server is shutting down count 228
error All 1 node(s) tried for the query failed (showing first 1 nodes, use getAllErrors() for more): Node(endPoint=/172.18.0.2:32146, hostId=35a8ed41-4a70-4945-9eec-53a299c03fb7, hashCode=683c0c99): [java.io.IOException: Connection reset by peer] count 9
error All 1 node(s) tried for the query failed (showing first 1 nodes, use getAllErrors() for more): Node(endPoint=/172.18.0.2:32146, hostId=35a8ed41-4a70-4945-9eec-53a299c03fb7, hashCode=683c0c99): [io.netty.channel.StacklessClosedChannelException] count 1
error All 1 node(s) tried for the query failed (showing first 1 nodes, use getAllErrors() for more): Node(endPoint=/172.18.0.2:32146, hostId=35a8ed41-4a70-4945-9eec-53a299c03fb7, hashCode=683c0c99): [com.datastax.oss.driver.api.core.NodeUnavailableException: No connection was available to Node(endPoint=/172.18.0.2:32146, hostId=35a8ed41-4a70-4945-9eec-53a299c03fb7, hashCode=683c0c99)] count 33394
##### 526346 scans, 95800 errors
error Node(endPoint=/172.18.0.2:32146, hostId=35a8ed41-4a70-4945-9eec-53a299c03fb7, hashCode=683c0c99) is overloaded: Server is shutting down count 228
error All 1 node(s) tried for the query failed (showing first 1 nodes, use getAllErrors() for more): Node(endPoint=/172.18.0.2:32146, hostId=35a8ed41-4a70-4945-9eec-53a299c03fb7, hashCode=683c0c99): [java.io.IOException: Connection reset by peer] count 9
error All 1 node(s) tried for the query failed (showing first 1 nodes, use getAllErrors() for more): Node(endPoint=/172.18.0.2:32146, hostId=35a8ed41-4a70-4945-9eec-53a299c03fb7, hashCode=683c0c99): [io.netty.channel.StacklessClosedChannelException] count 1
error All 1 node(s) tried for the query failed (showing first 1 nodes, use getAllErrors() for more): Node(endPoint=/172.18.0.2:32146, hostId=35a8ed41-4a70-4945-9eec-53a299c03fb7, hashCode=683c0c99): [com.datastax.oss.driver.api.core.NodeUnavailableException: No connection was available to Node(endPoint=/172.18.0.2:32146, hostId=35a8ed41-4a70-4945-9eec-53a299c03fb7, hashCode=683c0c99)] count 95562
##### 537125 scans, 95800 errors
error Node(endPoint=/172.18.0.2:32146, hostId=35a8ed41-4a70-4945-9eec-53a299c03fb7, hashCode=683c0c99) is overloaded: Server is shutting down count 228
error All 1 node(s) tried for the query failed (showing first 1 nodes, use getAllErrors() for more): Node(endPoint=/172.18.0.2:32146, hostId=35a8ed41-4a70-4945-9eec-53a299c03fb7, hashCode=683c0c99): [java.io.IOException: Connection reset by peer] count 9
error All 1 node(s) tried for the query failed (showing first 1 nodes, use getAllErrors() for more): Node(endPoint=/172.18.0.2:32146, hostId=35a8ed41-4a70-4945-9eec-53a299c03fb7, hashCode=683c0c99): [io.netty.channel.StacklessClosedChannelException] count 1
error All 1 node(s) tried for the query failed (showing first 1 nodes, use getAllErrors() for more): Node(endPoint=/172.18.0.2:32146, hostId=35a8ed41-4a70-4945-9eec-53a299c03fb7, hashCode=683c0c99): [com.datastax.oss.driver.api.core.NodeUnavailableException: No connection was available to Node(endPoint=/172.18.0.2:32146, hostId=35a8ed41-4a70-4945-9eec-53a299c03fb7, hashCode=683c0c99)] count 95562
##### 547382 scans, 95800 errors
...
##### 1587961 scans, 95800 errors
...

After short-time period of errors during node shutdown reading continues gracefully during node reconnect.

[dkropachev]

It happens because attempts are storred on query struct, if you change your code from:

	for i := 0; i <= workers; i++ {
		wg.Add(1)

		go func() {
			defer wg.Done()
			query := session.Query("insert into k8stest.test (a, b) values (?,?)")
			for j := i * queries; j < (i+1)*queries; j++ {
				query.Bind(j, "Message"+strconv.Itoa(j))
				if err := query.Exec(); err != nil {
					panic(err)
				}
			}
			query.Release()
		}()
	}

To:

	for i := 0; i <= workers; i++ {
		wg.Add(1)

		go func() {
			defer wg.Done()
			for j := i * queries; j < (i+1)*queries; j++ {
				query := session.Query("insert into k8stest.test (a, b) values (?,?)")
				query.Bind(j, "Message"+strconv.Itoa(j))
				if err := query.Exec(); err != nil {
					panic(err)
				}
			}
			query.Release()
		}()
	}

It should work.
I am not saying it is right, but it just how it is.

[dkropachev]

Fix: #1885

[isopov]

I test errors during reads not during inserts. And it seems that I am doing a very similar thing:

query = session.Query("select b from k8stest.test where a=?")

i recreate query from the session on error.

[dkropachev]

I test errors during reads not during inserts. And it seems that I am doing a very similar thing:

query = session.Query("select b from k8stest.test where a=?")

i recreate query from the session on error.

Can you please checkout fixed version and see if you have this issue there.
On my setup with your code I am not getting any errors.

[isopov]

Can you please checkout fixed version and see if you have this issue there.

I used the version from your PR

replace github.com/gocql/gocql => github.com/go-auxiliaries/gocql v0.0.0-20250424171856-e62d4d23dc59

Behavior is the same. Depending on what cassandra node I shutdown there may be zero errors during its shutdown. But when it comes back online session breaks and I receive only "no hosts available in the pool" errors after its join.

Here is another perspective on this problems. Code is the same. However when it starts reading from cassandra I add another node to the ring. I do it with

helm upgrade --namespace cassandra cassandra bitnami/cassandra --set "replicaCount=7"

Somewhere near final stages of new node joining the ring session breaks and starts returning "no hosts available in the pool" for all queries.

So the problem is not in the host going down, but in the host going up.

[isopov]

What is common with this breakage on adding new cassandra node and restarting existing node is that I receive 101 errors error gocql: connection closed waiting for response count. And seems to be always exactly 101 such errors.

[joao-r-reis]

Are you able by chance to reproduce this by using a simpler approach with docker/docker-compose or ccm? I'm interested in trying to reproduce this but I don't have a k8s environment ready to use atm

[isopov]

Note: I've tried reproducing it with ccm and could not do this. I believe this issue is somewhere in communication with k8s service.

[joao-r-reis]

If you're using domain names as contact points maybe the underlying IPs are changing and gocql is not updating them because it only resolves DNS once during session init?

[jameshartig]

@isopov are you still having this issue? Can you share what the system.peers table looks like for a given node in the cluster? The driver keeps a list of all peers and keeps a pool of connections for each peer (that aren't filtered out by a filter function). As peers are removed and replaced, the driver will make new pools to the new peers. It doesn't seem like that is happening which makes me think the data in the system.peers table is inaccurate.