Token-aware Load Balancing Policy

Token-aware policy is used to reduce network hops whenever possible by sending requests directly to the node that owns the data. Token-aware policy acts as a filter, wrapping another load balancing policy.

Token-aware policy uses schema metadata available in the cluster to determine the right partitioners and replication strategies for a given keyspace and locate replicas for a given statement.

In case replica node(s) cannot be found or reached, this policy falls back onto the wrapped policy plan.

Background

Given

a running cassandra cluster in 2 datacenters with 2 nodes in each

And

the following schema:

CREATE KEYSPACE simplex WITH replication = {'class': 'NetworkTopologyStrategy', 'dc1': 1, 'dc2': 1};
CREATE TABLE simplex.songs (
  id uuid PRIMARY KEY,
  title text,
  album text,
  artist text,
  tags set<text>,
  data blob
);
INSERT INTO simplex.songs (id, title, album, artist, tags)
VALUES (
   756716f7-2e54-4715-9f00-91dcbea6cf50,
   'La Petite Tonkinoise',
   'Bye Bye Blackbird',
   'Joséphine Baker',
   {'jazz', '2013'})
;
INSERT INTO simplex.songs (id, title, album, artist, tags)
VALUES (
   f6071e72-48ec-4fcb-bf3e-379c8a696488,
   'Die Mösch',
   'In Gold',
   'Willi Ostermann',
   {'kölsch', '1996', 'birds'}
);
INSERT INTO simplex.songs (id, title, album, artist, tags)
VALUES (
   fbdf82ed-0063-4796-9c7c-a3d4f47b4b25,
   'Memo From Turner',
   'Performance',
   'Mick Jager',
   {'soundtrack', '1991'}
);
CREATE TABLE simplex.playlists (
  id uuid,
  title text,
  album text,
  artist text,
  song_id uuid,
  PRIMARY KEY ((id, title), album, artist)
);
INSERT INTO simplex.playlists (id, song_id, title, album, artist)
VALUES (
   2cc9ccb7-6221-4ccb-8387-f22b6a1b354d,
   756716f7-2e54-4715-9f00-91dcbea6cf50,
   'La Petite Tonkinoise',
   'Bye Bye Blackbird',
   'Joséphine Baker'
);
INSERT INTO simplex.playlists (id, song_id, title, album, artist)
VALUES (
   2cc9ccb7-6221-4ccb-8387-f22b6a1b354d,
   f6071e72-48ec-4fcb-bf3e-379c8a696488,
   'Die Mösch',
   'In Gold',
   'Willi Ostermann'
);
INSERT INTO simplex.playlists (id, song_id, title, album, artist)
VALUES (
   3fd2bedf-a8c8-455a-a462-0cd3a4353c54,
   fbdf82ed-0063-4796-9c7c-a3d4f47b4b25,
   'Memo From Turner',
   'Performance',
   'Mick Jager'
);
INSERT INTO simplex.playlists (id, song_id, title, album, artist)
VALUES (
   3fd2bedf-a8c8-455a-a462-0cd3a4353c54,
   756716f7-2e54-4715-9f00-91dcbea6cf50,
   'La Petite Tonkinoise',
   'Bye Bye Blackbird',
   'Joséphine Baker'
);

Requests are routed to the primary replica

Given

the following example:

require 'cassandra'

base_policy = Cassandra::LoadBalancing::Policies::DCAwareRoundRobin.new('dc2')
policy      = Cassandra::LoadBalancing::Policies::TokenAware.new(base_policy)
cluster     = Cassandra.cluster(load_balancing_policy: policy)
session     = cluster.connect('simplex')
statement   = session.prepare("SELECT token(id) FROM songs WHERE id = ?")

[
  Cassandra::Uuid.new('756716f7-2e54-4715-9f00-91dcbea6cf50'),
  Cassandra::Uuid.new('f6071e72-48ec-4fcb-bf3e-379c8a696488'),
  Cassandra::Uuid.new('fbdf82ed-0063-4796-9c7c-a3d4f47b4b25')
].each do |uuid|
  result  = session.execute(statement, arguments: [uuid])
  replica = result.execution_info.hosts.first
  total   = result.execution_info.hosts.size
  puts "uuid=#{uuid} token=#{result.first.values.first} replica=#{replica.ip} total=#{total}"
end

When

it is executed

Then

its output should contain:

uuid=756716f7-2e54-4715-9f00-91dcbea6cf50 token=-4565826248849633211 replica=127.0.0.4 total=1
uuid=f6071e72-48ec-4fcb-bf3e-379c8a696488 token=-1176857621403111796 replica=127.0.0.4 total=1
uuid=fbdf82ed-0063-4796-9c7c-a3d4f47b4b25 token=2440231132048646025 replica=127.0.0.3 total=1

Requests are routed according to wrapped policy plan when primary replica is down

Given

the following example:

require 'cassandra'

base_policy = Cassandra::LoadBalancing::Policies::DCAwareRoundRobin.new('dc2')
policy      = Cassandra::LoadBalancing::Policies::TokenAware.new(base_policy)
cluster     = Cassandra.cluster(load_balancing_policy: policy)
session     = cluster.connect('simplex')
statement   = session.prepare("SELECT token(id) FROM songs WHERE id = ?")

[
  Cassandra::Uuid.new('756716f7-2e54-4715-9f00-91dcbea6cf50'),
  Cassandra::Uuid.new('f6071e72-48ec-4fcb-bf3e-379c8a696488'),
  Cassandra::Uuid.new('fbdf82ed-0063-4796-9c7c-a3d4f47b4b25')
].each do |uuid|
  result  = session.execute(statement, arguments: [uuid], consistency: :one)
  replica = result.execution_info.hosts.first
  total   = result.execution_info.hosts.size
  puts "uuid=#{uuid} token=#{result.first.values.first} replica=#{replica.ip} total=#{total}"
end

And

node 4 is stopped

When

it is executed

Then

its output should contain:

uuid=756716f7-2e54-4715-9f00-91dcbea6cf50 token=-4565826248849633211 replica=127.0.0.3 total=1
uuid=f6071e72-48ec-4fcb-bf3e-379c8a696488 token=-1176857621403111796 replica=127.0.0.3 total=1
uuid=fbdf82ed-0063-4796-9c7c-a3d4f47b4b25 token=2440231132048646025 replica=127.0.0.3 total=1

Requests with compound partition keys are routed to the primary replica

Given

the following example:

require 'cassandra'

policy    = Cassandra::LoadBalancing::Policies::DCAwareRoundRobin.new('dc2')
policy    = Cassandra::LoadBalancing::Policies::TokenAware.new(policy)
cluster   = Cassandra.cluster(load_balancing_policy: policy)
session   = cluster.connect('simplex')
statement = session.prepare("SELECT token(id, title) FROM playlists WHERE id = ? AND title = ?")

[
  [Cassandra::Uuid.new('2cc9ccb7-6221-4ccb-8387-f22b6a1b354d'), 'La Petite Tonkinoise'],
  [Cassandra::Uuid.new('2cc9ccb7-6221-4ccb-8387-f22b6a1b354d'), 'Die Mösch'],
  [Cassandra::Uuid.new('3fd2bedf-a8c8-455a-a462-0cd3a4353c54'), 'Memo From Turner'],
  [Cassandra::Uuid.new('3fd2bedf-a8c8-455a-a462-0cd3a4353c54'), 'La Petite Tonkinoise'],
].each do |arguments|
  result  = session.execute(statement, arguments: arguments)
  replica = result.execution_info.hosts.first
  total   = result.execution_info.hosts.size
  puts "uuid=#{arguments[0]} title=#{arguments[0]} token=#{result.first.values.first} replica=#{replica.ip} total=#{total}"
end

When

it is executed

Then

its output should contain:

uuid=2cc9ccb7-6221-4ccb-8387-f22b6a1b354d title=2cc9ccb7-6221-4ccb-8387-f22b6a1b354d token=6231549073425362204 replica=127.0.0.3 total=1
uuid=2cc9ccb7-6221-4ccb-8387-f22b6a1b354d title=2cc9ccb7-6221-4ccb-8387-f22b6a1b354d token=-115815985718975675 replica=127.0.0.4 total=1
uuid=3fd2bedf-a8c8-455a-a462-0cd3a4353c54 title=3fd2bedf-a8c8-455a-a462-0cd3a4353c54 token=-463065628644986368 replica=127.0.0.4 total=1
uuid=3fd2bedf-a8c8-455a-a462-0cd3a4353c54 title=3fd2bedf-a8c8-455a-a462-0cd3a4353c54 token=-8087998491924709995 replica=127.0.0.4 total=1

Requests with local consistencies are routed to a remote datacenter when primary replica is down

Given

the following example:

require 'cassandra'

max_remote_hosts_to_use = nil
use_remote_hosts_for_local_consistency = true
base_policy = Cassandra::LoadBalancing::Policies::DCAwareRoundRobin.new('dc2',
                                                                         max_remote_hosts_to_use,
                                                                         use_remote_hosts_for_local_consistency)
policy      = Cassandra::LoadBalancing::Policies::TokenAware.new(base_policy)
cluster     = Cassandra.cluster(load_balancing_policy: policy)
session     = cluster.connect('simplex')
statement   = session.prepare("SELECT token(id) FROM songs WHERE id = ?")

[
  Cassandra::Uuid.new('756716f7-2e54-4715-9f00-91dcbea6cf50'),
  Cassandra::Uuid.new('f6071e72-48ec-4fcb-bf3e-379c8a696488')
].each do |uuid|
  result  = session.execute(statement, arguments: [uuid], consistency: :local_one)
  replica = result.execution_info.hosts.first
  total   = result.execution_info.hosts.size
  puts "uuid=#{uuid} token=#{result.first.values.first} replica=#{replica.ip} total=#{total}"
end

And

node 3 is stopped

And

node 4 is stopped

When

it is executed

Then

its output should contain:

uuid=756716f7-2e54-4715-9f00-91dcbea6cf50 token=-4565826248849633211 replica=127.0.0.2 total=1
uuid=f6071e72-48ec-4fcb-bf3e-379c8a696488 token=-1176857621403111796 replica=127.0.0.1 total=1