Load balancing¶

Node distance¶

For each node, the policy computes a distance that determines how connections will be established:

• LOCAL and REMOTE are “active” distances, meaning that the driver will keep open connections to this node. Connection pools can be sized independently for each distance.

• IGNORED means that the driver will never attempt to connect.

Typically, the distance will reflect network topology (e.g. local vs. remote datacenter), although that is entirely up to each policy implementation. It can also change over time.

Query plan¶

Each time the driver executes a query, it asks the policy to compute a query plan, in other words a list of nodes. The driver then tries each node in sequence, moving down the plan according to the retry policy and speculative execution policy.

The contents and order of query plans are entirely implementation-specific, but policies typically return plans that:

• are different for each query, in order to balance the load across the cluster;

• only contain nodes that are known to be able to process queries, i.e. neither ignored nor down;

• favor local nodes over remote ones.

Local only¶

The default policy only connects to a single datacenter. The rationale is that a typical multi-region deployment will collocate one or more application instances with each Cassandra datacenter:

                  /----+----\
                  | client  |
                  \----+----/
                       |
                       v
               /---------------\
               | load balancer |
               \-------+-------/
                       |
          +------------+------------+
          |                         |
+---------|---------+     +---------|---------+
| Region1 v         |     | Region2 v         |
|    /---------\    |     |    /---------\    |
|    |   app1  |    |     |    |   app2  |    |
|    \----+----/    |     |    \----+----/    |
|         |         |     |         |         |
|         v         |     |         v         |
|   +-----------+   |     |   +-----------+   |
|   | {s}       |   |     |   | {s}       |   |
|   | Cassandra +------=------+ Cassandra |   |
|   |    DC1    |   |     |   |    DC2    |   |
|   +-----------+   |     |   +-----------+   |
|                   |     |                   |
+-------------------+     +-------------------+

In previous driver versions, you could configure application-level failover, such as: “if all the Cassandra nodes in DC1 are down, allow app1 to connect to the nodes in DC2”. We now believe that this is not the right place to handle this: if a whole datacenter went down at once, it probably means a catastrophic failure happened in Region1, and the application node is down as well. Failover should be cross-region instead (handled by the load balancer in this example).

Therefore the default policy does not allow remote nodes; it only ever assigns the LOCAL or IGNORED distance. You must provide a local datacenter name, either in the configuration:

datastax-java-driver.basic.load-balancing-policy {
  local-datacenter = datacenter1
}

Or programmatically when building the session:

CqlSession session = CqlSession.builder()
    .withLocalDatacenter("datacenter1")
    .build();

If both are provided, the programmatic value takes precedence.

For convenience, the local datacenter name may be omitted if no contact points were provided: in that case, the driver will connect to 127.0.0.1:9042, and use that node’s datacenter. This is just for a better out-of-the-box experience for users who have just downloaded the driver; beyond that initial development phase, you should provide explicit contact points and a local datacenter.

Finding the local datacenter¶

To check which datacenters are defined in a given cluster, you can run nodetool status. It will print information about each node in the cluster, grouped by datacenters. Here is an example:

$ nodetool status
Datacenter: DC1
===============
Status=Up/Down
|/ State=Normal/Leaving/Joining/Moving
--  Address    Load       Tokens  Owns  Host ID  Rack
UN  <IP1>      1.5 TB     256     ?     <ID1>    rack1
UN  <IP2>      1.5 TB     256     ?     <ID2>    rack2
UN  <IP3>      1.5 TB     256     ?     <ID3>    rack3

Datacenter: DC2
===============
Status=Up/Down
|/ State=Normal/Leaving/Joining/Moving
--  Address    Load       Tokens  Owns  Host ID  Rack
UN  <IP4>      1.5 TB     256     ?     <ID4>    rack1
UN  <IP5>      1.5 TB     256     ?     <ID5>    rack2
UN  <IP6>      1.5 TB     256     ?     <ID6>    rack3

To find out which datacenter should be considered local, you need to first determine which nodes the driver is going to be co-located with, then choose their datacenter as local. In case of doubt, you can also use cqlsh; if cqlsh is co-located too in the same datacenter, simply run the command below:

cqlsh> select data_center from system.local;

data_center
-------------
DC1 

Token-aware¶

The default policy is token-aware by default: requests will be routed in priority to the replicas that own the data being queried.

CREATE TABLE testKs.sensor_data(id int, year int, ts timestamp, data double,
                                PRIMARY KEY ((id, year), ts));

SimpleStatement statement =
    SimpleStatement.newInstance(
        "SELECT * FROM testKs.sensor_data WHERE id = 1 and year = 2016");

// No routing info available:
assert statement.getRoutingKeyspace() == null;
assert statement.getRoutingKey() == null;

// Set the keyspace manually (skip this if using a per-query keyspace):
statement = statement.setRoutingKeyspace("testKs");

// Set the routing key manually: serialize each partition key component to its target CQL type
statement = statement.setRoutingKey(
    TypeCodecs.INT.encodePrimitive(1, session.getContext().getProtocolVersion()),
    TypeCodecs.INT.encodePrimitive(2016, session.getContext().getProtocolVersion()));

session.execute(statement);

// All components bound: all info available
PreparedStatement pst1 =
    session.prepare("SELECT * FROM testKs.sensor_data WHERE id = :id and year = :year");
BoundStatement statement1 = pst1.bind(1, 2016);

assert statement1.getRoutingKeyspace() != null;
assert statement1.getRoutingKey() != null;

// 'id' hard-coded, only 'year' is bound: only keyspace available
PreparedStatement pst2 =
    session.prepare("SELECT * FROM testKs.sensor_data WHERE id = 1 and year = :year");
BoundStatement statement2 = pst2.bind(2016);

assert statement2.getRoutingKeyspace() != null;
assert statement2.getRoutingKey() == null;

Optional node filtering¶

Finally, the default policy accepts an optional node filter that gets applied just after the test for inclusion in the local DC. If a node doesn’t pass this test, it will be set at distance IGNORED and the driver will never try to connect to it. This is a good way to exclude nodes on some custom criteria.

You can pass the filter through the configuration:

datastax-java-driver.basic.load-balancing-policy {
  class = DefaultLoadBalancingPolicy
  local-datacenter = datacenter1
  filter-class = com.acme.MyNodeFilter
}

The filter class must implement java.util.function.predicate<Node>, and have a public constructor that takes a DriverContext argument: public MyNodeFilter(DriverContext context).

Sometimes it’s more convenient to pass the filter programmatically; you can do that with SessionBuilder.withNodeFilter:

List<Node> whiteList = ...
CqlSession session = CqlSession.builder()
    .withNodeFilter(whiteList::contains)
    .build();

If a programmatic filter is provided, the configuration option is ignored.