Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 2 additions & 6 deletions doc/modules/cassandra/pages/architecture/dynamo.adoc
Original file line number Diff line number Diff line change
Expand Up @@ -245,12 +245,8 @@ data. Configure it by specifying replication factor as
`<total_replicas>/<transient_replicas` Both `SimpleStrategy` and
`NetworkTopologyStrategy` support configuring transient replication.

Transiently replicated keyspaces only support tables created with
`read_repair` set to `NONE`; monotonic reads are not currently
supported. You also can't use `LWT`, logged batches, or counters in {40_version}.
You will possibly never be able to use materialized views with
transiently replicated keyspaces and probably never be able to use
secondary indices with them.
Transient replication requires mutation tracking, so a transiently replicated
keyspace must set `replication_type = 'tracked'`.

Transient replication is an experimental feature that is not ready
for production use. The expected audience is experienced users of
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -71,8 +71,8 @@ CREATE KEYSPACE some_keysopace WITH replication = {'class': 'NetworkTopologyStra
'DC1' : '3/1'', 'DC2' : '5/2'};
....

Transiently replicated keyspaces only support tables with `read_repair`
set to `NONE`.
Transient replication requires mutation tracking, so a transiently replicated
keyspace must set `replication_type = 'tracked'`.

Important Restrictions:

Expand All @@ -83,15 +83,78 @@ must first remove all transient replicas, then change the # of full
replicas, then add back the transient replicas.
* You can only safely increase number of transients one at a time with
incremental repair run in between each time.
* You can't add transient replicas to a keyspace whose mutation tracking
migration is still in progress, nor enable mutation tracking and add
transient replicas in the same statement. Set `replication_type = 'tracked'`
first, wait for the migration to complete, then alter the replication
factor.
* You can't promote a transient replica to a full replica, because it holds
no data for the range it witnessed. Either drop the transient replica from
the replica set (`3/1` to `2`), or add a full replica before removing it
(`3/1` to `4/1` to `3`) with a full repair run in between if the replica
count must be preserved.

Setting `-Dcassandra.allow_unsafe_transient_changes=true` skips every restriction
above except the first, which is governed by
`-Dcassandra.allow_alter_rf_during_range_movement`. Any alteration reached that
way can leave a transient replica counted by reads it cannot answer, and needs a
full repair afterwards to redistribute the data.

Promotion is the one performed deliberately often enough to want it without a
restart, so it also honours `AllowUnsafeWitnessPromotion` on the
`StorageService` MBean, seeded from
`-Dcassandra.allow_unsafe_witness_promotion`.

=== Adding transient replicas to an existing keyspace

A keyspace that is not already using mutation tracking adopts transient
replication in two steps, because enabling mutation tracking starts a migration
and reads for a range that has not yet migrated still take the untracked path.

First enable mutation tracking:

....
ALTER KEYSPACE some_keyspace WITH replication_type = 'tracked';
....

The migration repairs each range so that a single-replica tracked read reflects
every write made while the keyspace was untracked. Wait for it to complete, then
set the replication factor:

....
ALTER KEYSPACE some_keyspace WITH replication = {'class': 'SimpleStrategy',
'replication_factor' : '3/1'};
....

A keyspace created with both settings at once needs no migration, because it has
no pre-tracking writes to repair.

=== Removing transient replicas

Dropping the transient replica from the replica set needs no data movement,
because the remaining full replicas already hold everything:

....
ALTER KEYSPACE some_keyspace WITH replication = {'class': 'SimpleStrategy',
'replication_factor' : '2'};
....

Where the range must keep the same number of full replicas, add one before
removing the transient replica, and run a full repair in between so the added
replica is populated:

....
ALTER KEYSPACE some_keyspace WITH replication = {'class': 'SimpleStrategy',
'replication_factor' : '4/1'};
nodetool repair -full some_keyspace
ALTER KEYSPACE some_keyspace WITH replication = {'class': 'SimpleStrategy',
'replication_factor' : '3'};
....

Additionally, transient replication cannot be used for:

* Monotonic Reads
* Lightweight Transactions (LWTs)
* Logged Batches
* Counters
* Keyspaces using materialized views
* Secondary indexes (2i)
* Logged Batches.
* Keyspaces using materialized views.

== Cheap Quorums

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ public enum CassandraRelevantProperties
ALLOW_UNSAFE_JOIN("cassandra.allow_unsafe_join"),
ALLOW_UNSAFE_REPLACE("cassandra.allow_unsafe_replace"),
ALLOW_UNSAFE_TRANSIENT_CHANGES("cassandra.allow_unsafe_transient_changes"),
ALLOW_UNSAFE_WITNESS_PROMOTION("cassandra.allow_unsafe_witness_promotion"),
APPROXIMATE_TIME_PRECISION_MS("cassandra.approximate_time_precision_ms", "2"),
ASYNC_PROFILER_ENABLED("cassandra.async_profiler.enabled", "false"),
ASYNC_PROFILER_UNSAFE_MODE("cassandra.async_profiler.unsafe_mode", "false"),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,15 @@
*/
package org.apache.cassandra.cql3.statements.schema;

import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import com.google.common.collect.Sets;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand All @@ -34,6 +39,7 @@
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.LocalStrategy;
import org.apache.cassandra.locator.NetworkTopologyStrategy;
import org.apache.cassandra.locator.ReplicationFactor;
import org.apache.cassandra.locator.SimpleStrategy;
import org.apache.cassandra.schema.KeyspaceMetadata;
Expand All @@ -42,7 +48,6 @@
import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.membership.NodeId;
Expand All @@ -53,6 +58,7 @@

import static org.apache.cassandra.config.CassandraRelevantProperties.ALLOW_ALTER_RF_DURING_RANGE_MOVEMENT;
import static org.apache.cassandra.config.CassandraRelevantProperties.ALLOW_UNSAFE_TRANSIENT_CHANGES;
import static org.apache.cassandra.config.CassandraRelevantProperties.ALLOW_UNSAFE_WITNESS_PROMOTION;
import static org.apache.cassandra.replication.MutationTrackingService.DISABLED_MESSAGE;

public final class AlterKeyspaceStatement extends AlterSchemaStatement
Expand Down Expand Up @@ -104,6 +110,7 @@ public Keyspaces apply(ClusterMetadata metadata)

validateNoRangeMovements();
validateTransientReplication(keyspace, newKeyspace);
validateWitnessTransitions(metadata, keyspace, newKeyspace);

// Because we used to not properly validate unrecognized options, we only log a warning if we find one.
try
Expand Down Expand Up @@ -225,10 +232,6 @@ private void validateTransientReplication(KeyspaceMetadata current, KeyspaceMeta

if (!current.views.isEmpty())
throw new ConfigurationException("Cannot use transient replication on keyspaces using materialized views");

for (TableMetadata table : current.tables)
if (!table.indexes.isEmpty())
throw new ConfigurationException("Cannot use transient replication on keyspaces using secondary indexes");
}

//This is true right now because the transition from transient -> full lacks the pending state
Expand All @@ -237,6 +240,12 @@ private void validateTransientReplication(KeyspaceMetadata current, KeyspaceMeta
if (oldFull > newFull && oldTrans > 0)
throw new ConfigurationException("Can't add full replicas if there are any transient replicas. You must first remove all transient replicas, then change the # of full replicas, then add back the transient replicas");

//Promoting a witness to a full replica leaves it a full replica holding no data for the range
//it witnessed: tracked writes skip the column family store for tokens outside a node's full
//ranges, see Keyspace#applyInternalTracked. The promoted replica is then counted by quorum
//reads. Repair has to redistribute the data, so this is opt-in only.
validateNoWitnessPromotion(current.replicationStrategy, proposed.replicationStrategy);

//Don't increase transient replication factor by more than one at a time if changing number of replicas
//Just like with changing full replicas it's not safe to do this as you could read from too many replicas
//that don't have the necessary data. W/O transient replication this alteration was allowed and it's not clear
Expand All @@ -247,6 +256,111 @@ private void validateTransientReplication(KeyspaceMetadata current, KeyspaceMeta
throw new ConfigurationException("Can only safely increase number of transients one at a time with incremental repair run in between each time");
}

/**
* A witness is promoted when the transient count falls while the full count rises. This promotes witness replicas
* to full replicas without sending them the data they need to participate in reads.
*/
private void validateNoWitnessPromotion(AbstractReplicationStrategy current, AbstractReplicationStrategy proposed)
{
if (allowUnsafeWitnessPromotion)
return;

if (!current.getClass().equals(proposed.getClass()))
{
if (promotesWitness(current.getReplicationFactor(), proposed.getReplicationFactor()))
throw witnessPromotionRejected();
return;
}

Map<String, ReplicationFactor> before = replicationFactorsByGroup(current);
Map<String, ReplicationFactor> after = replicationFactorsByGroup(proposed);
for (String group : Sets.union(before.keySet(), after.keySet()))
{
if (promotesWitness(before.getOrDefault(group, ReplicationFactor.ZERO),
after.getOrDefault(group, ReplicationFactor.ZERO)))
throw witnessPromotionRejected();
}
}

private static boolean promotesWitness(ReplicationFactor before, ReplicationFactor after)
{
return after.transientReplicas() < before.transientReplicas() && after.fullReplicas > before.fullReplicas;
}

/**
* Replication factors keyed by the unit the strategy assigns them to independently: datacenter for
* {@link NetworkTopologyStrategy}, and the keyspace as a whole for every other strategy, whose
* aggregate factor is already its only group.
*/
private static Map<String, ReplicationFactor> replicationFactorsByGroup(AbstractReplicationStrategy strategy)
{
if (!(strategy instanceof NetworkTopologyStrategy))
return Collections.singletonMap("", strategy.getReplicationFactor());

NetworkTopologyStrategy nts = (NetworkTopologyStrategy) strategy;
Map<String, ReplicationFactor> byDc = new HashMap<>();
for (String dc : nts.getDatacenters())
byDc.put(dc, nts.getReplicationFactor(dc));
return byDc;
}

private ConfigurationException witnessPromotionRejected()
{
return new ConfigurationException(String.format("Cannot promote a transient replica of %s to a full replica: " +
"it holds no data for the range it witnessed. Set %s=true over " +
"JMX, or %s=true at startup, to allow it, then run a full repair " +
"to distribute the data.",
keyspaceName,
ALLOW_UNSAFE_WITNESS_PROMOTION.getKey(),
ALLOW_UNSAFE_TRANSIENT_CHANGES.getKey()));
}

private static volatile boolean allowUnsafeWitnessPromotion = ALLOW_UNSAFE_WITNESS_PROMOTION.getBoolean();

public static void setAllowUnsafeWitnessPromotion(boolean allow)
{
allowUnsafeWitnessPromotion = allow;
}

public static boolean getAllowUnsafeWitnessPromotion()
{
return allowUnsafeWitnessPromotion;
}

/**
* Alterations that would leave a witness (transient replica) counted by reads it cannot answer.
* These are unsafe rather than inexpressible, so cassandra.allow_unsafe_transient_changes opts out
* of them alongside the rules in validateTransientReplication.
*/
private void validateWitnessTransitions(ClusterMetadata metadata,
KeyspaceMetadata current,
KeyspaceMetadata proposed)
{
if (allow_unsafe_transient_changes)
return;

boolean addsWitnesses = proposed.replicationStrategy.getReplicationFactor().hasTransientReplicas();

// A migrating keyspace has pending ranges, and reads for a pending range take the untracked
// path, see MigrationRouter#shouldUseTrackedForReads. Those reads would contact a transient
// replica, which RangeCommandIterator#executeNormal rejects outright. Migrating a pending range
// also relies on blocking read repair to converge the replicas, and a witness cannot take part.
if (addsWitnesses && metadata.mutationTrackingMigrationState.isMigrating(keyspaceName))
throw new ConfigurationException(String.format("Cannot add transient replicas to %s while its mutation " +
"tracking migration is in progress. Wait for the migration " +
"to complete, then alter the replication factor.",
keyspaceName));

// AlterSchema#maybeUpdateMutationTrackingMigrationState starts the migration after this
// statement validates, so the check above cannot see one this statement is about to start.
if (addsWitnesses && proposed.params.replicationType.isTracked() && !current.params.replicationType.isTracked())
throw new ConfigurationException(String.format("Cannot enable mutation tracking on %s and add transient " +
"replicas in the same statement, because doing so starts a " +
"migration. Set replication_type = 'tracked' first, wait for " +
"the migration to complete, then alter the replication factor.",
keyspaceName));
}

@Override
public AuditLogContext getAuditLogContext()
{
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,6 @@
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.consensus.TransactionalMode;
import org.apache.cassandra.service.consensus.migration.TransactionalMigrationFromMode;
import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.membership.Directory;
Expand Down Expand Up @@ -700,12 +699,6 @@ public KeyspaceMetadata apply(Epoch epoch, KeyspaceMetadata keyspace, TableMetad
"before being replayed.");
}

if (keyspace.replicationStrategy.hasTransientReplicas()
&& params.readRepair != ReadRepairStrategy.NONE)
{
throw ire("read_repair must be set to 'NONE' for transiently replicated keyspaces");
}

if (!params.compression.isEnabled())
Guardrails.uncompressedTablesEnabled.ensureEnabled(state);

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,6 @@
import org.apache.cassandra.schema.UserFunctions;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.ClientWarn;
import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.transport.Event.SchemaChange;
Expand Down Expand Up @@ -194,12 +193,6 @@ public Keyspaces apply(ClusterMetadata metadata)
if (sourceTableMeta.isCompactTable())
Guardrails.compactTablesEnabled.ensureEnabled(state);

if (sourceKeyspaceMeta.replicationStrategy.hasTransientReplicas()
&& sourceTableMeta.params.readRepair != ReadRepairStrategy.NONE)
{
throw ire("read_repair must be set to 'NONE' for transiently replicated keyspaces");
}

if (!sourceTableMeta.params.compression.isEnabled())
Guardrails.uncompressedTablesEnabled.ensureEnabled(state);

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,6 @@ public final class CreateIndexStatement extends AlterSchemaStatement
public static final String TABLE_DOES_NOT_EXIST = "Table '%s' doesn't exist";
public static final String COUNTER_TABLES_NOT_SUPPORTED = "Secondary indexes on counter tables aren't supported";
public static final String MATERIALIZED_VIEWS_NOT_SUPPORTED = "Secondary indexes on materialized views aren't supported";
public static final String TRANSIENTLY_REPLICATED_KEYSPACE_NOT_SUPPORTED = "Secondary indexes are not supported on transiently replicated keyspaces";
public static final String CUSTOM_CREATE_WITHOUT_COLUMN = "Only CUSTOM indexes can be created without specifying a target column";
public static final String CUSTOM_MULTIPLE_COLUMNS = "Only CUSTOM indexes support multiple columns";
public static final String DUPLICATE_TARGET_COLUMN = "Duplicate column '%s' in index target list";
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -175,13 +175,6 @@ public Keyspaces apply(ClusterMetadata metadata)
TableMetadata table = builder.build();
table.validate();

// TODO (review): This can be removed right? ReadRepair is effectively not done anymore so the setting doesn't matter
// if (keyspace.replicationStrategy.hasTransientReplicas()
// && table.params.readRepair != ReadRepairStrategy.NONE)
// {
// throw ire("read_repair must be set to 'NONE' for transiently replicated keyspaces");
// }

if (!table.params.compression.isEnabled() && !SchemaConstants.isSystemKeyspace(table.keyspace))
Guardrails.uncompressedTablesEnabled.ensureEnabled(state);

Expand Down
4 changes: 4 additions & 0 deletions src/java/org/apache/cassandra/locator/ReplicaPlans.java
Original file line number Diff line number Diff line change
Expand Up @@ -234,6 +234,10 @@ public static Replica findCounterLeaderReplica(ClusterMetadata metadata, String

EndpointsForToken replicas = metadata.placements.get(keyspace.getMetadata().params.replication).reads.forToken(key.getToken()).get();

// A transient replica holds no data for the range it witnesses, so it would resolve against nothing
// and write a value that discards every prior increment.
replicas = replicas.filter(Replica::isFull);

// CASSANDRA-13043: filter out those endpoints not accepting clients yet, maybe because still bootstrapping
replicas = replicas.filter(replica -> StorageService.instance.isRpcReady(replica.endpoint()));

Expand Down
Loading