Java driver test using SCassandra.

gistfile1.java

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import com.codahale.metrics.ConsoleReporter;
import com.google.common.collect.ImmutableMap;
import com.google.common.util.concurrent.MoreExecutors;
import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;
import org.scassandra.http.client.PrimingClient;
import org.scassandra.http.client.PrimingRequest;
import org.scassandra.junit.ScassandraServerRule;

import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Metrics;
import com.datastax.driver.core.Session;

public class FixedLoadTest {
    // Set up a fake Cassandra host
    @ClassRule
    public static final ScassandraServerRule SCASSANDRA = new ScassandraServerRule();

    @Rule
    public final ScassandraServerRule resetScassandra = SCASSANDRA;

    private static final PrimingClient primingClient = SCASSANDRA.primingClient();

    @Test
    public void test() throws InterruptedException {
        // Simulate a long response time to avoid unnecessary pressure on client
        primingClient.prime(
            PrimingRequest.queryBuilder()
                .withQuery("select foo from bar")
                .withRows(ImmutableMap.of("bar", 1))
                .withFixedDelay(10 * 1000)
                .build()
        );

        Cluster cluster = null;
        try {
            cluster = Cluster.builder()
                .addContactPoint("127.0.0.1")
                .withPort(8042)
                .build();

            Metrics metrics = cluster.getMetrics();
            ConsoleReporter reporter = ConsoleReporter.forRegistry(metrics.getRegistry())
                .filter((name, metric) -> name.equals("open-connections"))
                .build();
            reporter.start(5, TimeUnit.SECONDS);

            Session session = cluster.connect();

            int wantedConcurrentRequests = 300;
            Semaphore semaphore = new Semaphore(wantedConcurrentRequests);
            while (true) {
                semaphore.acquire();
                session.executeAsync("select foo from bar")
                    .addListener(semaphore::release, MoreExecutors.sameThreadExecutor());
            }
        } finally {
            if (cluster != null)
                cluster.close();
        }
    }
}