NoHostAvailable Exception on Async operations with Datastax Cassandra driver

Question

I'm using nested Asynchronous query execution with Cassandra. Data is continuously streamed in and for each incoming data, the below block of cassandra operations are executed. It works fine for a while but then starts throwing a lot of NoHostAvailableException.
Please me help me out here. Cassandra Session Connection code :

I use separate sessions for read and write. Each of these sessions connect to a different seed as I was told this would improve performance.

final com.datastax.driver.core.Session readSession = CassandraManager.connect("10.22.1.144", "fr_repo",
                    "READ");
            final com.datastax.driver.core.Session writeSession = CassandraManager.connect("10.1.12.236", "fr_repo",
                    "WRITE");

The CassandraManager.connect method is below :

public static  Session connect(String ip, String keySpace,String type) {
           PoolingOptions poolingOpts = new PoolingOptions();
           poolingOpts.setCoreConnectionsPerHost(HostDistance.REMOTE, 2);
           poolingOpts.setMaxConnectionsPerHost(HostDistance.REMOTE, 400);
           poolingOpts.setMaxSimultaneousRequestsPerConnectionThreshold(HostDistance.REMOTE, 128);
           poolingOpts.setMinSimultaneousRequestsPerConnectionThreshold(HostDistance.REMOTE, 2);
           cluster = Cluster
             .builder()
             .withPoolingOptions( poolingOpts )
             .addContactPoint(ip)
             .withRetryPolicy( DowngradingConsistencyRetryPolicy.INSTANCE )
             .withReconnectionPolicy( new ConstantReconnectionPolicy( 100L ) ).build();
           Session s =  cluster.connect(keySpace);
           return s;
    }

Database operation code :

ResultSetFuture resultSetFuture = readSession.executeAsync(selectBound.bind(fr.getHashcode()));
                    Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
                        public void onSuccess(com.datastax.driver.core.ResultSet resultSet) {
                            try {
                                Iterator<Row> rows = resultSet.iterator();
                                if (!rows.hasNext()) {
                                    ResultSetFuture resultSetFuture = readSession.executeAsync(selectPrimaryBound
                                            .bind(fr.getPrimaryKeyHashcode()));
                                    Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
                                        public void onFailure(Throwable arg0) {

                                        }

                                        public void onSuccess(ResultSet arg0) {
                                            Iterator<Row> rows = arg0.iterator();
                                            if (!rows.hasNext()) {
                                                writeSession.executeAsync(insertBound.bind(fr.getHashcode(), fr,
                                                        System.currentTimeMillis()));
                                                    writeSession.executeAsync(insertPrimaryBound.bind(
                                                            fr.getHashcode(),
                                                            fr.getCombinedPrimaryKeys(), System.currentTimeMillis()));
                                                produceintoQueue(new Gson().toJson(frCompleteMap));

                                            } else {
                                                writeSession.executeAsync(updateBound.bind(fr,
                                                        System.currentTimeMillis(), fr.getHashcode()));
                                                produceintoQueue(new Gson().toJson(frCompleteMap));
                                            }
                                        }

                                    });

                                } else {
                                    writeSession.executeAsync(updateLastSeenBound.bind(System.currentTimeMillis(),
                                        fr.getHashcode()));

                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                            }

                        }

Answer 1

It sounds like you're sending more requests than your pool/cluster can handle. This is pretty easy to do when you're never actually waiting for a result, as is the case in your code. You're essentially just throwing as many requests as you can into the pipeline with no blocking, and there's no natural back pressure to slow down your app if the pool or cluster get backed up. So if your request volume is too high, eventually all the hosts will be busy with the backed up work queue. You can use nodetool tpstats to see what your request queues look like on each node.