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.

The driver built-in policies only ever assign the LOCAL or IGNORED distance, to avoid cross- datacenter traffic (see below to understand how to change this behavior).

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.

Datacenter locality¶

By default, both DefaultLoadBalancingPolicy and DcInferringLoadBalancingPolicy only connect 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    |   |
|   +-----------+   |     |   +-----------+   |
|                   |     |                   |
+-------------------+     +-------------------+

When using these policies 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.

datastax-java-driver.basic.load-balancing-policy {
  local-rack = rack1
}

$ 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

cqlsh> select data_center,rack from system.local;

 data_center | rack
-------------+-------
 datacenter1 | rack1

Cross-datacenter failover¶

Since the driver by default only contacts nodes in the local datacenter, what happens if the whole datacenter is down? Resuming the example shown in the diagram above, shouldn’t the driver temporarily allow app1 to connect to the nodes in DC2?

We believe that, while appealing by its simplicity, such ability is not the right way to handle a datacenter failure: resuming our example above, if the whole DC1 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 the above example).

However, due to popular demand, starting with driver 4.10, we re-introduced cross-datacenter failover in the driver built-in load balancing policies.

Cross-datacenter failover is enabled with the following configuration option:

datastax-java-driver.advanced.load-balancing-policy.dc-failover {
  max-nodes-per-remote-dc = 2
}

The default for max-nodes-per-remote-dc is zero, which means that failover is disabled. Setting this option to any value greater than zero will have the following effects:

• The load balancing policies will assign the REMOTE distance to that many nodes in each remote datacenter.

• The driver will then attempt to open connections to those nodes. The actual number of connections to open to each one of those nodes is configurable, see Connection pools for more details. By default, the driver opens only one connection to each node.

• Those remote nodes (and only those) will then become eligible for inclusion in query plans, effectively enabling cross-datacenter failover.

Beware that enabling such failover can result in cross-datacenter network traffic spikes, if the local datacenter is down or experiencing high latencies!

Cross-datacenter failover can also have unexpected consequences when using local consistency levels (LOCAL_ONE, LOCAL_QUORUM and LOCAL_SERIAL). Indeed, a local consistency level may have different semantics depending on the replication factor (RF) in use in each datacenter: if the local DC has RF=3 for a given keyspace, but the remote DC has RF=1 for it, achieving LOCAL_QUORUM in the local DC means 2 replicas required, but in the remote DC, only one will be required.

For this reason, cross-datacenter failover for local consistency levels is disabled by default. If you want to enable this and understand the consequences, then set the following option to true:

datastax-java-driver.advanced.load-balancing-policy.dc-failover {
  allow-for-local-consistency-levels = true
}

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;

Customizing node distance assignment¶

Finally, all the driver the built-in policies accept an optional node distance evaluator that gets invoked each time a node is added to the cluster or comes back up. If the evaluator returns a non-null distance for the node, that distance will be used, otherwise the driver will use its built-in logic to assign a default distance to it. This is a good way to exclude nodes or to adjust their distance according to custom, dynamic criteria.

You can pass the node distance evaluator through the configuration:

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

The node distance evaluator class must implement NodeDistanceEvaluator, and have a public constructor that takes a DriverContext argument: public MyNodeDistanceEvaluator(DriverContext context).

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

Map<Node, NodeDistance> distances = ...
CqlSession session = CqlSession.builder()
    .withNodeDistanceEvaluator((node, dc) -> distances.get(node))
    .build();

If a programmatic node distance evaluator evaluator is provided, the configuration option is ignored.