Cassandra BATCH architecture, in depth: mutations, the batchlog, replay, guardrails and timeouts

From statements to mutations

The coordinator first turns the batch's statements into mutations. A Cassandra mutation is all the changes to one partition key within one keyspace. Statements that hit the same key in the same keyspace, even in different tables, merge into one mutation; statements that hit different keys produce separate mutations. The number of mutations, not the number of statements, decides the path.

In code the decision is one line: atomic mutation happens only when the batch is logged and there is more than one mutation. That is why a logged batch confined to one partition costs no more than an unlogged one. Two corrections to common lore follow. First, merging across tables does not make the tables a single isolated unit; the documented isolation guarantee is per partition of one table. Second, the size guardrails below skip single-mutation batches entirely, so a huge single-partition batch is not warned about, even though it can still produce a huge partition.

From statements to mutations

The coordinator first turns the batch's statements into mutations. A Cassandra mutation is all the changes to one partition key within one keyspace. Statements that hit the same key in the same keyspace, even in different tables, merge into one mutation; statements that hit different keys produce separate mutations. The number of mutations, not the number of statements, decides the path.

In code the decision is one line: atomic mutation happens only when the batch is logged and there is more than one mutation. That is why a logged batch confined to one partition costs no more than an unlogged one. Two corrections to common lore follow. First, merging across tables does not make the tables a single isolated unit; the documented isolation guarantee is per partition of one table. Second, the size guardrails below skip single-mutation batches entirely, so a huge single-partition batch is not warned about, even though it can still produce a huge partition.

The logged path, step by step

For a logged batch with several mutations the coordinator runs mutateAtomically, in this order:

1. Availability check. It builds a response handler for every mutation at the requested consistency level. If any partition cannot reach that level with live replicas, the batch fails with UnavailableException before anything is written.
2. Choose batchlog endpoints. It picks up to two live nodes in the local datacenter, from different racks where possible, and uses itself only when the datacenter has a single node. The batchlog_endpoint_strategy setting controls the choice; the default, random_remote, avoids the local rack when it can.
3. Write the batchlog, synchronously. The whole batch is serialised and stored in system.batches on those endpoints, acknowledged at ONE if only one endpoint was found and TWO otherwise. Until this succeeds, no data mutation has been sent.
4. Send the mutations. Each mutation goes to its own replicas, and the coordinator waits for every one to reach the client's consistency level.
5. Remove the batchlog entry, asynchronously. Once all mutations have been acknowledged, a cleanup deletes the entry on the batchlog nodes. The client is answered without waiting for the removal.

The cost is visible in this list: one extra synchronous round trip to two nodes, the full batch payload written twice more (and later deleted, which creates tombstones in system.batches), and a coordinator that holds every mutation in memory until the slowest replica answers.

Replay: how atomicity is delivered

Every node runs a BatchlogManager that wakes every 10 seconds and scans its local system.batches for entries older than the batchlog timeout. By default that timeout is twice write_request_timeout, long enough for a healthy batch to finish and be removed; a system property can override it. An entry that old means the coordinator never confirmed the batch, usually because it crashed or a mutation timed out. The manager replays the stored mutations to their replicas, writing hints for replicas that are down, then deletes the entry.

Replay is throttled by batchlog_replay_throttle (1024KiB per second by default in 5.0, reduced in proportion to the number of nodes), so a large backlog drains slowly rather than overwhelming the cluster. Replay re-applies mutations with their original timestamps, so applying one twice is harmless. Two consequences matter operationally. A replayed batch can land seconds or minutes after the client gave up, so the system must tolerate late arrival. And if a table has gc_grace_seconds of zero, Cassandra warns on every logged batch that touches it, because a late replay could interact badly with tombstones that were already purged. Before deleting replayed entries, the manager flushes and fsyncs any hints it wrote, so a node crash during replay cannot lose the batch.

Guardrails and metrics

The coordinator applies three guardrails from cassandra.yaml, all measured in mutation data size and all skipped when the batch has only one mutation.

Do not raise the fail threshold to make an error go away; it exists because a large batch pins coordinator memory and, if logged, is written three times. For metrics, the Batch metrics group exposes histograms PartitionsPerLoggedBatch, PartitionsPerUnloggedBatch and PartitionsPerCounterBatch. A rising high percentile on the unlogged histogram is almost always an application using batches for bulk loading. A growing system.batches (check with nodetool tablestats system.batches) means replays are happening, which means coordinators are failing mid-batch.

Timeouts and what they tell the client

A WriteTimeoutException on a logged batch carries a write type, and the two values mean different things.

• BATCH_LOG: the batchlog write did not get its acknowledgements in time, so the coordinator never started the data mutations. That does not prove nothing will happen: one batchlog node may have stored the entry, and it will replay after the timeout window. Treat the outcome as unknown; retrying is safe when the statements are idempotent, because a retry and a replay then write the same values.
• BATCH: the batchlog was stored, and some mutation missed the consistency level. The batch will be applied by replay. Retrying is safe if the statements are idempotent, and often unnecessary.
• UNLOGGED_BATCH: some mutations may have been applied and others not, with no replay to finish the job. Retry only if every statement is idempotent.

Idempotency here means the statement produces the same row whatever it is applied on top of: an insert or update of fixed values is idempotent; a counter increment or a list append is not. The Java driver 4.x consults its retry policy on write timeouts only for statements marked idempotent, and the default policy then retries only when the write type is BATCH_LOG, and only once. Mark statements idempotent so that retry and speculative execution can act at all.

A related subtlety: the client's consistency level applies to the data mutations, not to the batchlog. The batchlog write always waits for its own one or two local endpoints, so even a batch at ONE pays that round trip, and both hops must fit inside one write timeout. Client code should handle the two logged outcomes separately:

BatchStatement batch = BatchStatement.builder(DefaultBatchType.LOGGED)
    .addStatement(insertUser.bind(userId, email, name))
    .addStatement(insertUserByEmail.bind(email, userId))
    .setIdempotent(true)                      // fixed values: safe to apply twice
    .setConsistencyLevel(DefaultConsistencyLevel.LOCAL_QUORUM)
    .build();
try {
  session.execute(batch);
} catch (WriteTimeoutException e) {
  if (e.getWriteType() == WriteType.BATCH) {
    // logged and stored: replay will finish it; record and move on
    metrics.counter("batch.timeout.replay_pending").increment();
  } else {
    throw e;  // BATCH_LOG after the driver's single retry: outcome unknown, surface it
  }
}

Worked example: a sign-up and a crashed coordinator

A sign-up writes a row to users keyed by user id and a row to users_by_email keyed by email: two keys, so two mutations, so a logged batch takes the batchlog path. Follow one failure.

1. The coordinator, node A, checks both partitions can reach LOCAL_QUORUM. They can.
2. It stores the batch on nodes C and E, in two other racks. Both acknowledge.
3. It sends the users mutation and the users_by_email mutation. The first reaches quorum. Before the second does, node A loses power.
4. The client sees a timeout from its connection to A, retries on node B, and the retry succeeds: both rows written with the same values.
5. About two write timeouts later, node C's batchlog manager finds the entry that A never removed, replays both mutations, and deletes it. The rows are rewritten with identical contents; nothing changes.

Had the application used an unlogged batch, or two separate writes, and not retried, the email lookup row would be missing until repair, and sign-in by email would fail for a user who exists. That missing lookup row is the failure logged batches exist to prevent. Note what the batch did not do: a reader querying users_by_email between steps 3 and 5 could see the user row without the lookup row. There is no isolation across partitions.

Bulk writes: what to do instead

For loading many rows, skip batches. An unlogged batch over many partitions sends every mutation through one coordinator, which then fans them out; the coordinator does the routing work a token-aware driver would have done, and one slow replica holds the whole response. Send individual asynchronous writes, routed by the driver to replicas, with a bound on how many are in flight.

Semaphore inFlight = new Semaphore(256);          // tune against coordinator latency
List<CompletableFuture<?>> pending = new ArrayList<>();
for (Row r : rows) {
  inFlight.acquireUninterruptibly();
  CompletionStage<AsyncResultSet> f = session.executeAsync(insert.bind(r.key(), r.value()));
  pending.add(f.toCompletableFuture().whenComplete((rs, err) -> {
    inFlight.release();
    if (err != null) failures.add(r);               // retry later if the insert is idempotent
  }));
}
CompletableFuture.allOf(pending.toArray(new CompletableFuture[0])).join();

The one batching that does help throughput is an unlogged batch of rows that share a partition key: it is one mutation, sent once, applied once.

Failure modes

• Bulk load through batches. Coordinator CPU and heap climb, latency spikes, size warnings flood the log. Switch to bounded async writes.
• Batchlog backlog. system.batches grows after node crashes or overload; replay is throttled, so it drains slowly. Find why coordinators fail mid-batch.
• Non-idempotent retry. A counter or list append in a retried batch is applied twice. Keep such statements out of batches that may be retried.
• Expecting isolation. Readers see half a multi-partition batch. Design reads to tolerate it, or keep the data in one partition.
• Mixed timestamps. Statements in one batch share a timestamp by default, so a delete and an insert of the same cell tie. The batch statements article covers the rule.

Trade-offs

Logged batches buy eventual all-or-nothing application for a small number of related partitions, paid for with an extra synchronous round trip, extra writes and tombstones, and coordinator memory. Use them for denormalised copies that must agree, such as a row and its lookup rows. Unlogged batches buy nothing across partitions except fewer client requests, at the coordinator's expense. When you need a read-modify-write guarantee rather than eventual completeness, the tool is a lightweight transaction, see Cassandra LWT: when and how, with its own costs. Consistency level choice for the mutations is explained in Cassandra consistency levels, and how replicas that were down catch up is in hinted handoff.

What to do next

1. Grep your code for batches and count distinct partition keys in each; classify every one using the table above.
2. Replace multi-partition unlogged batches with bounded asynchronous writes.
3. Keep logged batches for small groups of rows that must agree, and mark them idempotent.
4. Handle WriteType.BATCH and BATCH_LOG timeouts differently in client code.
5. Alert on the batch size warning in server logs and on the high percentile of partitions per unlogged batch.
6. Watch system.batches size; a sustained non-zero count means coordinators are failing mid-batch.
7. Leave batch_size_fail_threshold at its default unless you can explain the extra coordinator memory.