This is an automated email from the ASF dual-hosted git repository.
bdeggleston pushed a commit to branch cep-45-mutation-tracking
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cep-45-mutation-tracking by
this push:
new 1ba23a9cd4 CEP-46: update witness replication validation restrictions
1ba23a9cd4 is described below
commit 1ba23a9cd4263a20a0977ab949f628119a019ce1
Author: Blake Eggleston <[email protected]>
AuthorDate: Tue Sep 8 15:33:12 2026 -0700
CEP-46: update witness replication validation restrictions
Patch by Blake Eggleston; Reviewed by Francisco Guerrero for CASSANDRA-21674
---
.../cassandra/pages/architecture/dynamo.adoc | 8 +-
.../managing/operating/transientreplication.adoc | 75 +++++-
.../config/CassandraRelevantProperties.java | 1 +
.../statements/schema/AlterKeyspaceStatement.java | 128 +++++++++-
.../statements/schema/AlterTableStatement.java | 7 -
.../cql3/statements/schema/CopyTableStatement.java | 7 -
.../statements/schema/CreateIndexStatement.java | 1 -
.../statements/schema/CreateTableStatement.java | 7 -
.../org/apache/cassandra/locator/ReplicaPlans.java | 4 +
.../cassandra/replication/TrackedWriteRequest.java | 9 +-
.../apache/cassandra/service/StorageService.java | 14 +
.../cassandra/service/StorageServiceMBean.java | 3 +
.../cassandra/service/paxos/PaxosPrepare.java | 38 +--
.../test/TrackedCounterForwardingTest.java | 2 +
.../test/TrackedCounterWitnessForwardingTest.java | 188 ++++++++++++++
.../service/paxos/PaxosWitnessDataNodeTest.java | 250 ++++++++++++++++++
...eyspaceStatementUnsafeTransientChangesTest.java | 95 +++++++
.../AlterKeyspaceWitnessTransitionTest.java | 284 +++++++++++++++++++++
.../AlterSchemaMutationTrackingTest.java | 103 ++++++++
19 files changed, 1162 insertions(+), 62 deletions(-)
diff --git a/doc/modules/cassandra/pages/architecture/dynamo.adoc
b/doc/modules/cassandra/pages/architecture/dynamo.adoc
index 551d5a3338..5da99becc3 100644
--- a/doc/modules/cassandra/pages/architecture/dynamo.adoc
+++ b/doc/modules/cassandra/pages/architecture/dynamo.adoc
@@ -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
diff --git
a/doc/modules/cassandra/pages/managing/operating/transientreplication.adoc
b/doc/modules/cassandra/pages/managing/operating/transientreplication.adoc
index 3a0bfaf1cb..c813a3ed9a 100644
--- a/doc/modules/cassandra/pages/managing/operating/transientreplication.adoc
+++ b/doc/modules/cassandra/pages/managing/operating/transientreplication.adoc
@@ -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:
@@ -83,15 +83,74 @@ 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.
+
+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 all data:
+
+....
+ALTER KEYSPACE some_keyspace WITH replication = {'class': 'SimpleStrategy',
+'replication_factor' : '2'};
+....
+
+To restore the original full replica count, add a full replica after you drop
the
+transient replica. Then run a full repair to populate the added replica:
+
+....
+ALTER KEYSPACE some_keyspace WITH replication = {'class': 'SimpleStrategy',
+'replication_factor' : '3'};
+nodetool repair -full some_keyspace
+....
+
+Direct promotion (`3/1` to `3`) is rejected by default. You can enable
+`AllowUnsafeWitnessPromotion` through Java Management Extensions (JMX) to allow
+it, and then run a full repair immediately.
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
diff --git
a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
index b0631d98ac..97dcc8d5ab 100644
--- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
+++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
@@ -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"),
diff --git
a/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java
b/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java
index 30faee655e..bb5c4487d4 100644
---
a/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java
+++
b/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java
@@ -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;
@@ -34,6 +39,7 @@ import org.apache.cassandra.exceptions.ConfigurationException;
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;
@@ -42,7 +48,6 @@ import org.apache.cassandra.schema.Keyspaces;
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;
@@ -53,6 +58,7 @@ import org.apache.cassandra.utils.FBUtilities;
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
@@ -102,9 +108,6 @@ public final class AlterKeyspaceStatement extends
AlterSchemaStatement
newKeyspace.params.validate(keyspaceName, state, metadata);
newKeyspace.replicationStrategy.validate(metadata);
- validateNoRangeMovements();
- validateTransientReplication(keyspace, newKeyspace);
-
// Because we used to not properly validate unrecognized options, we
only log a warning if we find one.
try
{
@@ -156,6 +159,11 @@ public final class AlterKeyspaceStatement extends
AlterSchemaStatement
if (!DatabaseDescriptor.getMutationTrackingEnabled())
throw ire(DISABLED_MESSAGE);
}
+
+ ClusterMetadata metadata = ClusterMetadata.current();
+ validateNoRangeMovements(metadata);
+ validateTransientReplication(keyspace, newKeyspace);
+ validateWitnessTransitions(metadata, keyspace, newKeyspace);
}
@Override
@@ -176,12 +184,11 @@ public final class AlterKeyspaceStatement extends
AlterSchemaStatement
return clientWarnings;
}
- private void validateNoRangeMovements()
+ private void validateNoRangeMovements(ClusterMetadata metadata)
{
if (allow_alter_rf_during_range_movement)
return;
- ClusterMetadata metadata = ClusterMetadata.current();
NodeId nodeId =
metadata.directory.peerId(FBUtilities.getBroadcastAddressAndPort());
Set<InetAddressAndPort> notNormalEndpoints =
metadata.directory.states.entrySet().stream().filter(e ->
!e.getKey().equals(nodeId)).filter(e -> {
switch (e.getValue())
@@ -225,10 +232,6 @@ public final class AlterKeyspaceStatement extends
AlterSchemaStatement
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
@@ -237,6 +240,12 @@ public final class AlterKeyspaceStatement extends
AlterSchemaStatement
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
@@ -247,6 +256,105 @@ public final class AlterKeyspaceStatement extends
AlterSchemaStatement
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 full count rises while transient
replicas exist. 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 before.transientReplicas() > 0 && 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 proposesWitnesses =
proposed.replicationStrategy.getReplicationFactor().hasTransientReplicas();
+
+ if (proposesWitnesses &&
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));
+
+ if (proposesWitnesses && 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()
{
diff --git
a/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java
b/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java
index cebc174e24..c544ad867e 100644
---
a/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java
+++
b/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java
@@ -71,7 +71,6 @@ import org.apache.cassandra.schema.Views;
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;
@@ -700,12 +699,6 @@ public abstract class AlterTableStatement extends
AlterSchemaStatement
"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);
diff --git
a/src/java/org/apache/cassandra/cql3/statements/schema/CopyTableStatement.java
b/src/java/org/apache/cassandra/cql3/statements/schema/CopyTableStatement.java
index 1250cb79ec..c92dfed4b3 100644
---
a/src/java/org/apache/cassandra/cql3/statements/schema/CopyTableStatement.java
+++
b/src/java/org/apache/cassandra/cql3/statements/schema/CopyTableStatement.java
@@ -55,7 +55,6 @@ import org.apache.cassandra.schema.Triggers;
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;
@@ -194,12 +193,6 @@ public final class CopyTableStatement extends
AlterSchemaStatement
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);
diff --git
a/src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java
b/src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java
index fc1c2343f1..31e9b920a1 100644
---
a/src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java
+++
b/src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java
@@ -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";
diff --git
a/src/java/org/apache/cassandra/cql3/statements/schema/CreateTableStatement.java
b/src/java/org/apache/cassandra/cql3/statements/schema/CreateTableStatement.java
index 99eeef4a58..7eca3ab992 100644
---
a/src/java/org/apache/cassandra/cql3/statements/schema/CreateTableStatement.java
+++
b/src/java/org/apache/cassandra/cql3/statements/schema/CreateTableStatement.java
@@ -175,13 +175,6 @@ public final class CreateTableStatement extends
AlterSchemaStatement
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);
diff --git a/src/java/org/apache/cassandra/locator/ReplicaPlans.java
b/src/java/org/apache/cassandra/locator/ReplicaPlans.java
index 01de6d4a5d..34bca5d18a 100644
--- a/src/java/org/apache/cassandra/locator/ReplicaPlans.java
+++ b/src/java/org/apache/cassandra/locator/ReplicaPlans.java
@@ -234,6 +234,10 @@ public class ReplicaPlans
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()));
diff --git a/src/java/org/apache/cassandra/replication/TrackedWriteRequest.java
b/src/java/org/apache/cassandra/replication/TrackedWriteRequest.java
index 9a5ab64157..1b0bb5856d 100644
--- a/src/java/org/apache/cassandra/replication/TrackedWriteRequest.java
+++ b/src/java/org/apache/cassandra/replication/TrackedWriteRequest.java
@@ -195,7 +195,14 @@ public class TrackedWriteRequest
ReplicaPlan.ForWrite plan = ReplicaPlans.forWrite(keyspace,
consistencyLevel, token, ReplicaPlans.writeAll);
AbstractReplicationStrategy rs = plan.replicationStrategy();
- if (plan.lookup(FBUtilities.getBroadcastAddressAndPort()) == null)
+ Replica localReplica =
plan.lookup(FBUtilities.getBroadcastAddressAndPort());
+
+ // A counter leader resolves the increment against its local data, so
we can't use transient replicas as data replicas.
+ boolean witnessForCounter = localReplica != null
+ && localReplica.isTransient()
+ && mutation instanceof CounterMutation;
+
+ if (localReplica == null || witnessForCounter)
{
logger.trace("Remote tracked request {} {}", mutation, plan);
writeMetrics.remoteRequests.mark();
diff --git a/src/java/org/apache/cassandra/service/StorageService.java
b/src/java/org/apache/cassandra/service/StorageService.java
index 15512ac981..ca036dbe33 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -98,6 +98,7 @@ import org.apache.cassandra.config.DataStorageSpec;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.QueryHandler;
import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.statements.schema.AlterKeyspaceStatement;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Keyspace;
@@ -5599,6 +5600,19 @@ public class StorageService extends
NotificationBroadcasterSupport implements IE
return PaxosRepair.getSkipPaxosRepairCompatibilityCheck();
}
+ @Override
+ public void setAllowUnsafeWitnessPromotion(boolean allow)
+ {
+ AlterKeyspaceStatement.setAllowUnsafeWitnessPromotion(allow);
+ logger.info("AllowUnsafeWitnessPromotion set to {} via jmx", allow);
+ }
+
+ @Override
+ public boolean getAllowUnsafeWitnessPromotion()
+ {
+ return AlterKeyspaceStatement.getAllowUnsafeWitnessPromotion();
+ }
+
@Override
public boolean topPartitionsEnabled()
{
diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java
b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
index 1e6b8e6a46..28e44de9a7 100644
--- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java
+++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
@@ -1366,6 +1366,9 @@ public interface StorageServiceMBean extends
NotificationEmitter
public void setSkipPaxosRepairCompatibilityCheck(boolean v);
public boolean getSkipPaxosRepairCompatibilityCheck();
+ public void setAllowUnsafeWitnessPromotion(boolean allow);
+ public boolean getAllowUnsafeWitnessPromotion();
+
String getToken(String keyspaceName, String table, String partitionKey);
public boolean topPartitionsEnabled();
public int getMaxTopSizePartitionCount();
diff --git a/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java
b/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java
index 97de0841e3..25c41f0c06 100644
--- a/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java
+++ b/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java
@@ -20,6 +20,7 @@ package org.apache.cassandra.service.paxos;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -31,6 +32,8 @@ import java.util.stream.Collectors;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+import com.google.common.annotations.VisibleForTesting;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -447,24 +450,10 @@ public class PaxosPrepare extends
PaxosRequestCallback<PaxosPrepare.Response> im
Message<R> selfMessage = null;
Message<R> summaryMessage = null;
Id readId = Id.nextId();
- Replica localReplica =
participants.lookup(FBUtilities.getBroadcastAddressAndPort());
- Replica dataNode = localReplica != null && localReplica.isFull()
- ? localReplica
- : null;
+ Replica dataNode = selectDataNode(participants,
FBUtilities.getBroadcastAddressAndPort());
int[] summaryHostIds = new int[participants.sizeOfPoll() - 1]; // all
nodes except data node
int summaryIndex = 0;
ClusterMetadata metadata = ClusterMetadata.current();
- if (dataNode == null)
- {
- for (int i = 0, size = participants.sizeOfPoll() ; i < size ; i++)
- {
- Replica replica = participants.voterReplica(i);
- if (!replica.isFull() ||
participants.electorate.isPending(replica.endpoint()))
- continue;
- dataNode = replica;
- break;
- }
- }
checkState(dataNode != null, "Couldn't find a data node to use");
int dataNodeId = metadata.directory.peerId(dataNode.endpoint()).id();
@@ -474,6 +463,8 @@ public class PaxosPrepare extends
PaxosRequestCallback<PaxosPrepare.Response> im
if (replica != dataNode &&
!participants.electorate.isPending(replica.endpoint()))
summaryHostIds[summaryIndex++] =
metadata.directory.peerId(replica.endpoint()).id();
}
+ if (summaryIndex < summaryHostIds.length)
+ summaryHostIds = Arrays.copyOf(summaryHostIds, summaryIndex);
for (int i = 0, size = participants.sizeOfPoll() ; i < size ; ++i)
{
@@ -506,6 +497,23 @@ public class PaxosPrepare extends
PaxosRequestCallback<PaxosPrepare.Response> im
}
}
+ @VisibleForTesting
+ static Replica selectDataNode(Participants participants,
InetAddressAndPort local)
+ {
+ Replica localReplica = participants.lookup(local);
+ if (localReplica != null && localReplica.isFull() &&
!participants.electorate.isPending(local))
+ return localReplica;
+
+ for (int i = 0, size = participants.sizeOfPoll() ; i < size ; i++)
+ {
+ Replica replica = participants.voterReplica(i);
+ if (!replica.isFull() ||
participants.electorate.isPending(replica.endpoint()))
+ continue;
+ return replica;
+ }
+ return null;
+ }
+
private static <R extends AbstractRequest<R>> void
startUntracked(PaxosPrepare prepare, Participants participants, Message<R>
send, BiFunction<R, RequestTime, Future<Response>> selfHandler)
{
prepare.haveTrackedDataResponseIfNeeded = true;
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/TrackedCounterForwardingTest.java
b/test/distributed/org/apache/cassandra/distributed/test/TrackedCounterForwardingTest.java
index 656b3be4f5..611491b939 100644
---
a/test/distributed/org/apache/cassandra/distributed/test/TrackedCounterForwardingTest.java
+++
b/test/distributed/org/apache/cassandra/distributed/test/TrackedCounterForwardingTest.java
@@ -35,6 +35,8 @@ import static
org.apache.cassandra.distributed.shared.AssertUtils.row;
*
* With 6 nodes and RF=3, we can pick a coordinator that is definitely
* not a replica for a specific partition.
+ *
+ * For a keyspace with witnesses, see {@link
TrackedCounterWitnessForwardingTest}.
*/
public class TrackedCounterForwardingTest extends TestBaseImpl
{
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/TrackedCounterWitnessForwardingTest.java
b/test/distributed/org/apache/cassandra/distributed/test/TrackedCounterWitnessForwardingTest.java
new file mode 100644
index 0000000000..6c9fef22a2
--- /dev/null
+++
b/test/distributed/org/apache/cassandra/distributed/test/TrackedCounterWitnessForwardingTest.java
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.locator.Replica;
+import org.apache.cassandra.locator.ReplicaPlans;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Counter writes on a keyspace with witnesses (transient replicas). The
witness-free equivalent is
+ * {@link TrackedCounterForwardingTest}.
+ *
+ * A counter leader resolves the increment against its local data
+ * ({@link org.apache.cassandra.db.CounterMutation#processModifications}) and
replicates the resolved
+ * absolute value. A witness holds no data for the range it witnesses, so a
witness elected leader resolves
+ * against nothing and writes a value discarding every prior increment.
+ *
+ * The proximity implementation sorts witnesses first, which is what
+ * {@link ReplicaPlans#findCounterLeaderReplica} falls back on when no replica
is in the local
+ * datacenter. See {@link
WitnessAlwaysReadsFullReplicaTest.TransientFirstProximity}.
+ */
+public class TrackedCounterWitnessForwardingTest extends TestBaseImpl
+{
+ private static final String TABLE = "counters";
+ private static final String QUALIFIED_TABLE = KEYSPACE + '.' + TABLE;
+ private static final String KEY = "1";
+
+ private static Cluster witnessCluster() throws IOException
+ {
+ Cluster cluster = Cluster.build(3)
+ .withConfig(cfg -> cfg.with(Feature.NETWORK)
+ .with(Feature.GOSSIP)
+ // NATIVE_PROTOCOL so
Instance startup reaches
+ //
CassandraDaemon#startNativeTransport, which sets
+ // rpc-readiness.
findCounterLeaderReplica filters on it.
+
.with(Feature.NATIVE_PROTOCOL)
+
.set("transient_replication_enabled", "true")
+ .set("dynamic_snitch",
false)
+ .set("node_proximity",
WitnessAlwaysReadsFullReplicaTest.TransientFirstProximity.class.getName()))
+ .start();
+ cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH
replication = " +
+ "{'class': 'SimpleStrategy',
'replication_factor': '3/1'} AND replication_type='tracked'");
+ cluster.schemaChange("CREATE TABLE " + QUALIFIED_TABLE + " (k text
PRIMARY KEY, c counter)");
+ return cluster;
+ }
+
+ @Test
+ public void testElectedLeaderIsAlwaysFull() throws IOException
+ {
+ try (Cluster cluster = witnessCluster())
+ {
+ // Election picks at random among candidates, so one election
proves nothing
+ int elections = 500;
+ int transientElections = cluster.get(1).callOnInstance(() -> {
+ ClusterMetadata metadata = ClusterMetadata.current();
+ DecoratedKey key = Schema.instance.getTableMetadata(KEYSPACE,
TABLE)
+
.partitioner.decorateKey(ByteBufferUtil.bytes(KEY));
+ String localDc =
DatabaseDescriptor.getLocator().local().datacenter;
+
+ int transientCount = 0;
+ for (int i = 0; i < elections; i++)
+ {
+ if (ReplicaPlans.findCounterLeaderReplica(metadata,
KEYSPACE, key, localDc, ConsistencyLevel.ONE).isTransient())
+ transientCount++;
+ }
+ return transientCount;
+ });
+
+ assertThat(transientElections)
+ .describedAs("a witness was elected counter leader; it holds
no data for the range it witnesses")
+ .isZero();
+ }
+ }
+
+ @Test
+ public void testIncrementsAreNotLost() throws IOException
+ {
+ try (Cluster cluster = witnessCluster())
+ {
+ int increments = 50;
+ for (int i = 0; i < increments; i++)
+ cluster.coordinator(1).execute("UPDATE " + QUALIFIED_TABLE + "
SET c = c + 1 WHERE k = ?",
+
org.apache.cassandra.distributed.api.ConsistencyLevel.QUORUM, KEY);
+
+ Object[][] result = cluster.coordinator(1).execute("SELECT c FROM
" + QUALIFIED_TABLE + " WHERE k = ?",
+
org.apache.cassandra.distributed.api.ConsistencyLevel.QUORUM, KEY);
+ assertThat(result).hasNumberOfRows(1);
+ assertThat((long) result[0][0])
+ .describedAs("an increment resolved against a witness's empty
data would reset the counter")
+ .isEqualTo(increments);
+ }
+ }
+
+ /**
+ * With every full replica for the token shut down, the write has no
candidate leader. It must fail
+ * rather than fall back on the witness, which would reset the counter to
the increment's own value.
+ */
+ @Test
+ public void testUnavailableWhenNoFullReplicaIsUp() throws IOException
+ {
+ try (Cluster cluster = witnessCluster())
+ {
+ cluster.coordinator(1).execute("UPDATE " + QUALIFIED_TABLE + " SET
c = c + 7 WHERE k = ?",
+
org.apache.cassandra.distributed.api.ConsistencyLevel.QUORUM, KEY);
+
+ List<Integer> fullReplicaNodes = fullReplicaNodesFor(cluster, KEY);
+ assertThat(fullReplicaNodes).describedAs("a 3/1 keyspace has two
full replicas").hasSize(2);
+ int coordinator = nodeOtherThan(cluster, fullReplicaNodes);
+
+ // Shut down rather than filtering messages: a message filter does
not convict the node in the
+ // failure detector, so a leader would still be elected and the
write would time out instead
+ for (int node : fullReplicaNodes)
+ cluster.get(node).shutdown().get();
+
+ assertThatThrownBy(() ->
+ cluster.coordinator(coordinator).execute("UPDATE " +
QUALIFIED_TABLE + " SET c = c + 1 WHERE k = ?",
+
org.apache.cassandra.distributed.api.ConsistencyLevel.ONE, KEY))
+ .describedAs("with no full replica up the counter write must
fail rather than be led by the witness")
+ .isInstanceOf(Throwable.class);
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static List<Integer> fullReplicaNodesFor(Cluster cluster, String
key)
+ {
+ String endpoints = cluster.get(1).callOnInstance(() -> {
+ ClusterMetadata metadata = ClusterMetadata.current();
+ Token token = Schema.instance.getTableMetadata(KEYSPACE, TABLE)
+
.partitioner.getToken(ByteBufferUtil.bytes(key));
+ StringBuilder sb = new StringBuilder();
+
metadata.placements.get(Schema.instance.getKeyspaceMetadata(KEYSPACE).params.replication)
+ .reads.forToken(token).get()
+ .stream()
+ .filter(Replica::isFull)
+ .forEach(r -> sb.append(sb.length() == 0 ? "" :
",").append(r.endpoint().getHostAddress(false)));
+ return sb.toString();
+ });
+
+ List<Integer> nodes = new ArrayList<>();
+ for (String endpoint : endpoints.split(","))
+
nodes.add(Integer.parseInt(endpoint.substring(endpoint.lastIndexOf('.') + 1)));
+ return nodes;
+ }
+
+ private static int nodeOtherThan(Cluster cluster, List<Integer> excluded)
+ {
+ for (int node = 1; node <= cluster.size(); node++)
+ if (!excluded.contains(node))
+ return node;
+ throw new AssertionError("no node available outside " + excluded);
+ }
+}
diff --git
a/test/unit/org/apache/cassandra/service/paxos/PaxosWitnessDataNodeTest.java
b/test/unit/org/apache/cassandra/service/paxos/PaxosWitnessDataNodeTest.java
new file mode 100644
index 0000000000..c69cd2be21
--- /dev/null
+++ b/test/unit/org/apache/cassandra/service/paxos/PaxosWitnessDataNodeTest.java
@@ -0,0 +1,250 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file to you under the
Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.paxos;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Predicate;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.ServerTestUtils;
+import org.apache.cassandra.config.CassandraRelevantProperties;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.Replica;
+import org.apache.cassandra.replication.MutationJournal;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.SchemaTestUtil;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.service.paxos.PaxosPrepare.selectDataNode;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+public class PaxosWitnessDataNodeTest
+{
+ private static final String KEYSPACE = "paxos_witness_test";
+ private static final String SINGLE_FULL_KEYSPACE =
"paxos_witness_single_full_test";
+ private static final String TABLE = "tbl";
+
+ private static InetAddressAndPort node1;
+ private static InetAddressAndPort node2;
+ private static InetAddressAndPort node3;
+
+ @BeforeClass
+ public static void setUpClass() throws Exception
+ {
+
CassandraRelevantProperties.PARTITIONER.setString(Murmur3Partitioner.class.getName());
+ ServerTestUtils.daemonInitialization();
+ // Not prepareServer(): it registers the local node on a random token,
and these tests place all
+ // three nodes deterministically so the replica set for the key is
known
+ ServerTestUtils.prepareServerNoRegister();
+ ServerTestUtils.markCMS();
+ MutationJournal.start();
+
+ node1 = FBUtilities.getBroadcastAddressAndPort();
+ node2 = ClusterMetadataTestHelper.addr(2);
+ node3 = ClusterMetadataTestHelper.addr(3);
+
+ ClusterMetadataTestHelper.register(node1, "datacenter1", "rack1");
+ ClusterMetadataTestHelper.register(node2, "datacenter1", "rack1");
+ ClusterMetadataTestHelper.register(node3, "datacenter1", "rack1");
+ ClusterMetadataTestHelper.join(node1, new
Murmur3Partitioner.LongToken(Long.MIN_VALUE / 2));
+ ClusterMetadataTestHelper.join(node2, new
Murmur3Partitioner.LongToken(0L));
+ ClusterMetadataTestHelper.join(node3, new
Murmur3Partitioner.LongToken(Long.MAX_VALUE / 2));
+
+ ClusterMetadataTestHelper.createKeyspace("CREATE KEYSPACE " + KEYSPACE
+
+ " WITH replication =
{'class': 'SimpleStrategy', 'replication_factor': '3/1'}" +
+ " AND replication_type =
'tracked'");
+ SchemaTestUtil.announceNewTable(TableMetadata.builder(KEYSPACE, TABLE)
+
.addPartitionKeyColumn("k", Int32Type.instance)
+ .addRegularColumn("v",
Int32Type.instance)
+ .build());
+
+ // A second keyspace with only one full replica. Killing it leaves a
quorum of witnesses, which is
+ // the only way to reach a live poll with no full replica in it:
replica ordering puts full
+ // replicas first, so a first-match selection would otherwise always
find one.
+ ClusterMetadataTestHelper.createKeyspace("CREATE KEYSPACE " +
SINGLE_FULL_KEYSPACE +
+ " WITH replication =
{'class': 'SimpleStrategy', 'replication_factor': '3/2'}" +
+ " AND replication_type =
'tracked'");
+
SchemaTestUtil.announceNewTable(TableMetadata.builder(SINGLE_FULL_KEYSPACE,
TABLE)
+
.addPartitionKeyColumn("k", Int32Type.instance)
+ .addRegularColumn("v",
Int32Type.instance)
+ .build());
+ }
+
+ private static Paxos.Participants participants(Predicate<Replica> isAlive)
+ {
+ return participants(KEYSPACE, ByteBufferUtil.bytes(1), isAlive);
+ }
+
+ private static Paxos.Participants participants(String keyspace,
Predicate<Replica> isAlive)
+ {
+ return participants(keyspace, ByteBufferUtil.bytes(1), isAlive);
+ }
+
+ private static Paxos.Participants participants(String keyspace, ByteBuffer
key, Predicate<Replica> isAlive)
+ {
+ TableMetadata table = Schema.instance.getTableMetadata(keyspace,
TABLE);
+ assertNotNull("test schema was not created", table);
+ Token token = table.partitioner.getToken(key);
+ return Paxos.Participants.get(ClusterMetadata.current(), table, token,
ConsistencyLevel.SERIAL, isAlive);
+ }
+
+
+ /**
+ * Replica ordering puts full replicas first, so a live poll normally
contains one whether or not the
+ * selection filters for it. With a single full replica and that replica
down, the surviving quorum is
+ * all witnesses, and the selection has to report that no data node is
available rather than picking a
+ * witness that cannot answer the read.
+ */
+ @Test
+ public void testNoDataNodeWhenOnlyWitnessesSurvive()
+ {
+ Paxos.Participants all = participants(SINGLE_FULL_KEYSPACE, replica ->
true);
+ assertEquals(1, fullReplicas(all).size());
+ InetAddressAndPort onlyFull = fullReplicas(all).get(0).endpoint();
+
+ Paxos.Participants participants = participants(SINGLE_FULL_KEYSPACE,
+ replica ->
!replica.endpoint().equals(onlyFull));
+
+ assertEquals(2, participants.sizeOfPoll());
+ assertEquals(participants.sizeOfPoll(),
witnesses(participants).size());
+
+ Replica dataNode = selectDataNode(participants,
participants.voterReplica(0).endpoint());
+ assertNull("startTracked fails the prepare with \"Couldn't find a data
node to use\" in this state",
+ dataNode);
+ }
+
+ private static List<Replica> fullReplicas(Paxos.Participants participants)
+ {
+ List<Replica> full = new ArrayList<>();
+ for (int i = 0, size = participants.sizeOfPoll(); i < size; i++)
+ if (participants.voterReplica(i).isFull())
+ full.add(participants.voterReplica(i));
+ return full;
+ }
+
+ private static List<Replica> witnesses(Paxos.Participants participants)
+ {
+ List<Replica> transientReplicas = new ArrayList<>();
+ for (int i = 0, size = participants.sizeOfPoll(); i < size; i++)
+ if (participants.voterReplica(i).isTransient())
+ transientReplicas.add(participants.voterReplica(i));
+ return transientReplicas;
+ }
+
+
+ @Test
+ public void testWitnessVotesButIsNotTheDataNode()
+ {
+ Paxos.Participants participants = participants(replica -> true);
+
+ assertEquals(3, participants.sizeOfPoll());
+ assertEquals(1, witnesses(participants).size());
+
+ Replica dataNode = selectDataNode(participants, node1);
+ assertNotNull(dataNode);
+ assertTrue(dataNode.isFull());
+ }
+
+ /**
+ * The case a cluster test cannot reach: with one full replica down,
quorum is still met by the
+ * surviving full replica plus the witness, so the prepare proceeds and
the data node has to be the
+ * surviving full replica rather than the witness.
+ */
+ @Test
+ public void testDataNodeIsTheSurvivingFullReplica()
+ {
+ Paxos.Participants all = participants(replica -> true);
+ Replica witness = witnesses(all).get(0);
+ Replica fullToKill = null;
+ for (int i = 0, size = all.sizeOfPoll(); i < size; i++)
+ {
+ Replica replica = all.voterReplica(i);
+ if (replica.isFull())
+ {
+ fullToKill = replica;
+ break;
+ }
+ }
+ assertNotNull(fullToKill);
+
+ InetAddressAndPort dead = fullToKill.endpoint();
+ Paxos.Participants participants = participants(replica ->
!replica.endpoint().equals(dead));
+
+ assertEquals(2, participants.sizeOfPoll());
+
+ // The coordinator here is the witness, so the local-replica shortcut
must not apply
+ Replica dataNode = selectDataNode(participants, witness.endpoint());
+ assertNotNull(dataNode);
+ assertTrue(dataNode.isFull());
+ assertNotEquals(dead, dataNode.endpoint());
+ }
+
+ @Test
+ public void testPendingNodeNotSelectedAsDataNode() throws Exception
+ {
+ InetAddressAndPort node4 = ClusterMetadataTestHelper.addr(4);
+ ClusterMetadataTestHelper.register(node4, "datacenter1", "rack1");
+ ClusterMetadataTestHelper.joinPartially(node4, new
Murmur3Partitioner.LongToken(Long.MAX_VALUE / 4));
+
+ try
+ {
+ TableMetadata table = Schema.instance.getTableMetadata(KEYSPACE,
TABLE);
+ Token token = new Murmur3Partitioner.LongToken(Long.MIN_VALUE / 4);
+ Paxos.Participants participants =
Paxos.Participants.get(ClusterMetadata.current(), table, token,
ConsistencyLevel.SERIAL, replica -> true);
+ assertTrue(participants.electorate.isPending(node1));
+
+ Replica dataNode = selectDataNode(participants, node1);
+ assertNotNull(dataNode);
+ assertTrue(dataNode.isFull());
+ assertNotEquals(node1, dataNode.endpoint());
+ }
+ finally
+ {
+
ClusterMetadataTestHelper.commit(ClusterMetadataTestHelper.getBootstrapPlan(node4).finishJoin);
+ }
+ }
+
+ /**
+ * Paxos ballots and promises live in system.paxos, a node-local table, so
being a witness for a data
+ * range does not affect a node's ability to hold Paxos state for it.
+ */
+ @Test
+ public void testPaxosStateKeyspaceIsNotTransientlyReplicated()
+ {
+ Keyspace systemKeyspace = Keyspace.open("system");
+
assertFalse(systemKeyspace.getReplicationStrategy().hasTransientReplicas());
+ }
+}
diff --git
a/test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceStatementUnsafeTransientChangesTest.java
b/test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceStatementUnsafeTransientChangesTest.java
new file mode 100644
index 0000000000..f33c8b37ad
--- /dev/null
+++
b/test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceStatementUnsafeTransientChangesTest.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file to you under the
Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tcm.transformations;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.ServerTestUtils;
+import org.apache.cassandra.config.CassandraRelevantProperties;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.replication.MutationJournal;
+import org.apache.cassandra.tcm.ClusterMetadata;
+
+import static
org.apache.cassandra.config.CassandraRelevantProperties.ALLOW_UNSAFE_TRANSIENT_CHANGES;
+import static org.apache.cassandra.cql3.CQLTester.schemaChange;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * The transient replication validation rules in AlterKeyspaceStatement are
skipped when
+ * cassandra.allow_unsafe_transient_changes is set, which is how an operator
opts into an unsafe
+ * alteration. The flag is read into a static final field at class
initialization, so it has to be set
+ * before AlterKeyspaceStatement loads. That is why these cases need their own
class rather than
+ * living in AlterSchemaMutationTrackingTest, which asserts the same
alterations are rejected.
+ */
+public class AlterKeyspaceStatementUnsafeTransientChangesTest
+{
+ private static final AtomicInteger ksCounter = new AtomicInteger();
+
+ @BeforeClass
+ public static void setUpClass() throws Exception
+ {
+ ALLOW_UNSAFE_TRANSIENT_CHANGES.setBoolean(true);
+
CassandraRelevantProperties.PARTITIONER.setString(Murmur3Partitioner.class.getName());
+ ServerTestUtils.daemonInitialization();
+ ServerTestUtils.prepareServer();
+ MutationJournal.start();
+ }
+
+ private static String nextKsName()
+ {
+ return "ks" + ksCounter.incrementAndGet();
+ }
+
+ @Test
+ public void testAddingWitnessesDuringMigrationAllowedWhenOptedIn()
+ {
+ String ksName = nextKsName();
+ schemaChange("CREATE KEYSPACE " + ksName +
+ " WITH replication = {'class': 'SimpleStrategy',
'replication_factor': '3'}" +
+ " AND replication_type = 'untracked'");
+ schemaChange(String.format("CREATE TABLE %s.tbl (pk int PRIMARY KEY,
val int)", ksName));
+ schemaChange(String.format("ALTER KEYSPACE %s WITH replication_type =
'tracked'", ksName));
+
assertTrue(ClusterMetadata.current().mutationTrackingMigrationState.isMigrating(ksName));
+
+ schemaChange("ALTER KEYSPACE " + ksName +
+ " WITH replication = {'class': 'SimpleStrategy',
'replication_factor': '3/1'}");
+
+ assertTrue(ClusterMetadata.current().schema.getKeyspaceMetadata(ksName)
+
.replicationStrategy.getReplicationFactor().hasTransientReplicas());
+ }
+
+ /** Enabling tracking and adding witnesses in a single statement. */
+ @Test
+ public void testEnablingTrackingWithWitnessesAllowedWhenOptedIn()
+ {
+ String ksName = nextKsName();
+ schemaChange("CREATE KEYSPACE " + ksName +
+ " WITH replication = {'class': 'SimpleStrategy',
'replication_factor': '3'}" +
+ " AND replication_type = 'untracked'");
+
+ schemaChange("ALTER KEYSPACE " + ksName +
+ " WITH replication = {'class': 'SimpleStrategy',
'replication_factor': '3/1'}" +
+ " AND replication_type = 'tracked'");
+
+ assertTrue(ClusterMetadata.current().schema.getKeyspaceMetadata(ksName)
+
.replicationStrategy.getReplicationFactor().hasTransientReplicas());
+ }
+}
diff --git
a/test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceWitnessTransitionTest.java
b/test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceWitnessTransitionTest.java
new file mode 100644
index 0000000000..55e2ad8291
--- /dev/null
+++
b/test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceWitnessTransitionTest.java
@@ -0,0 +1,284 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file to you under the
Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tcm.transformations;
+
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import org.apache.cassandra.ServerTestUtils;
+import org.apache.cassandra.config.CassandraRelevantProperties;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.replication.MutationJournal;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.schema.ReplicationParams;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.membership.Location;
+
+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.cql3.CQLTester.schemaChange;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Which replication changes a keyspace with witnesses (transient replicas)
permits.
+ *
+ * Each rejected transition asserts the whole rejection message rather than
just that the statement
+ * failed. Several rules overlap here -- witness promotion, the migration
guard, the pre-existing
+ * replica-count rules, and the requirement that transient replication implies
mutation tracking -- so a
+ * test asserting only failure can pass because the wrong rule fired.
+ *
+ * Datacenter-aware rows are present because {@link
org.apache.cassandra.locator.NetworkTopologyStrategy}
+ * reports the sum of its per-datacenter factors: comparing aggregates both
misses a promotion in one
+ * datacenter offset by a reduction in another, and rejects two independently
legal per-datacenter
+ * changes whose aggregates resemble a promotion.
+ */
+@RunWith(Parameterized.class)
+public class AlterKeyspaceWitnessTransitionTest
+{
+ private static final String SIMPLE_3 = "{'class': 'SimpleStrategy',
'replication_factor': '3'}";
+ private static final String SIMPLE_3_1 = "{'class': 'SimpleStrategy',
'replication_factor': '3/1'}";
+ private static final String NTS_3_1_AND_3 = "{'class':
'NetworkTopologyStrategy', 'DC1': '3/1', 'DC2': '3'}";
+ private static final String NTS_3_1_AND_2 = "{'class':
'NetworkTopologyStrategy', 'DC1': '3/1', 'DC2': '2'}";
+
+ private static final AtomicInteger ksCounter = new AtomicInteger();
+
+ private static final String TRACKED = "tracked";
+ private static final String UNTRACKED = "untracked";
+ private static final boolean MIGRATING = true;
+ private static final boolean NO_MIGRATION = false;
+
+ @Parameterized.Parameter(0)
+ public String startReplication;
+
+ /** The replication type the keyspace is created with. */
+ @Parameterized.Parameter(1)
+ public String startReplicationType;
+
+ /** Whether to leave a mutation tracking migration in flight before the
alteration under test. */
+ @Parameterized.Parameter(2)
+ public boolean startMigration;
+
+ @Parameterized.Parameter(3)
+ public String proposedReplication;
+
+ /** The replication type to propose, or null to leave it unchanged. */
+ @Parameterized.Parameter(4)
+ public String proposedReplicationType;
+
+ /** The rejection this change must produce, or null when it is permitted.
*/
+ @Parameterized.Parameter(5)
+ public Rejection expectedRejection;
+
+ @Parameterized.Parameter(6)
+ public String description;
+
+ enum Rejection
+ {
+ WITNESS_PROMOTION("Cannot promote a transient replica of %s to a full
replica: it holds no data for the " +
+ "range it witnessed. Set " +
ALLOW_UNSAFE_WITNESS_PROMOTION.getKey() + "=true over JMX, " +
+ "or " + ALLOW_UNSAFE_TRANSIENT_CHANGES.getKey() +
"=true at startup, to allow it, then " +
+ "run a full repair to distribute the data."),
+ MIGRATION_IN_FLIGHT("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."),
+ WOULD_START_MIGRATION("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."),
+ NEEDS_MUTATION_TRACKING("Transient replication requires mutation
tracking"),
+ REPLICA_COUNT_RULE("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");
+
+ private final String template;
+
+ Rejection(String template)
+ {
+ this.template = template;
+ }
+
+ String message(String keyspace)
+ {
+ return template.contains("%s") ? String.format(template, keyspace)
: template;
+ }
+ }
+
+ @Parameterized.Parameters(name = "{6}")
+ public static Collection<Object[]> transitions()
+ {
+ List<Object[]> rows = new ArrayList<>();
+
+ // The migration guard is on the migration, not on transient replicas
as such
+ rows.add(row(SIMPLE_3, TRACKED, NO_MIGRATION, "3/1", null, null,
+ "settled tracked keyspace accepts witnesses"));
+ rows.add(row(SIMPLE_3, UNTRACKED, MIGRATING, "3/1", null,
Rejection.MIGRATION_IN_FLIGHT,
+ "adding witnesses while a migration is in flight"));
+ rows.add(row(SIMPLE_3, UNTRACKED, NO_MIGRATION, "3/1", TRACKED,
Rejection.WOULD_START_MIGRATION,
+ "enabling tracking and adding witnesses at once"));
+
+ // Dropping a witness needs no data movement: replica ordering puts
the transient replica last, so
+ // lowering the factor removes it from the replica set rather than
promoting it
+ rows.add(row(SIMPLE_3_1, TRACKED, NO_MIGRATION, "2", null, null,
"dropping a witness"));
+ rows.add(row(SIMPLE_3_1, TRACKED, NO_MIGRATION, "2", UNTRACKED, null,
+ "dropping a witness and tracking at once"));
+ rows.add(row(SIMPLE_3, TRACKED, NO_MIGRATION, "3", UNTRACKED, null,
+ "leaving tracked replication without witnesses"));
+ rows.add(row(SIMPLE_3_1, TRACKED, NO_MIGRATION, "3/1", UNTRACKED,
Rejection.NEEDS_MUTATION_TRACKING,
+ "retaining witnesses while leaving tracked replication"));
+
+ // A promoted witness holds no data for the range it witnessed, and
quorum reads would count it
+ rows.add(row(SIMPLE_3_1, TRACKED, NO_MIGRATION, "3", null,
Rejection.WITNESS_PROMOTION,
+ "promoting a witness to a full replica"));
+ rows.add(row(SIMPLE_3_1, TRACKED, NO_MIGRATION, "3", UNTRACKED,
Rejection.WITNESS_PROMOTION,
+ "promoting a witness while leaving tracked replication"));
+ rows.add(row(SIMPLE_3_1, TRACKED, NO_MIGRATION, "4/1", null,
Rejection.WITNESS_PROMOTION,
+ "increasing full replicas while retaining a witness"));
+
+ // Per-datacenter comparison: the aggregate full replica count is 5
either side of this change
+ rows.add(row(NTS_3_1_AND_3, TRACKED, NO_MIGRATION,
+ "{'class': 'NetworkTopologyStrategy', 'DC1': '3', 'DC2':
'2'}", null,
+ Rejection.WITNESS_PROMOTION, "promotion masked by a
reduction in another datacenter"));
+ rows.add(row(NTS_3_1_AND_3, TRACKED, NO_MIGRATION,
+ "{'class': 'NetworkTopologyStrategy', 'DC1': '3', 'DC2':
'3'}", null,
+ Rejection.WITNESS_PROMOTION, "promotion in a single
datacenter"));
+ // Dropping a witness in one datacenter and adding a full replica in
another are legal alone; their
+ // aggregates resemble a promotion
+ rows.add(row(NTS_3_1_AND_2, TRACKED, NO_MIGRATION,
+ "{'class': 'NetworkTopologyStrategy', 'DC1': '2', 'DC2':
'3'}", null, null,
+ "legal changes in separate datacenters"));
+ // Not the promotion guard: removing a datacenter drops the aggregate
full count while a transient
+ // replica remains, which the pre-existing replica-count rule rejects
+ rows.add(row(NTS_3_1_AND_3, TRACKED, NO_MIGRATION,
+ "{'class': 'NetworkTopologyStrategy', 'DC2': '3'}", null,
+ Rejection.REPLICA_COUNT_RULE, "removing a witness
datacenter entirely"));
+
+ return rows;
+ }
+
+ private static Object[] row(String startReplication, String startType,
boolean startMigration,
+ String proposedReplication, String
proposedType, Rejection rejection,
+ String description)
+ {
+ return new Object[]{ startReplication, startType, startMigration,
proposedReplication, proposedType,
+ rejection, description };
+ }
+
+ @BeforeClass
+ public static void setUpClass() throws Exception
+ {
+
CassandraRelevantProperties.PARTITIONER.setString(Murmur3Partitioner.class.getName());
+ ServerTestUtils.daemonInitialization();
+ ServerTestUtils.prepareServer();
+ MutationJournal.start();
+
+ // Datacenter-aware rows need endpoints in both datacenters for their
options to be recognised
+ addEndpoints(new byte[]{ 10, 0, 0 }, new Location("DC1", "Rack1"));
+ addEndpoints(new byte[]{ 10, 20, 114 }, new Location("DC2", "Rack1"));
+ }
+
+ private static void addEndpoints(byte[] prefix, Location location) throws
UnknownHostException
+ {
+ for (byte last = 10; last < 14; last++)
+ {
+ InetAddressAndPort addr = InetAddressAndPort.getByAddress(new
byte[]{ prefix[0], prefix[1], prefix[2], last });
+ ClusterMetadataTestHelper.addEndpoint(addr,
Murmur3Partitioner.instance.getRandomToken(), location);
+ }
+ }
+
+ @Test
+ public void testTransition()
+ {
+ String ksName = "ks" + ksCounter.incrementAndGet();
+ schemaChange("CREATE KEYSPACE " + ksName + " WITH replication = " +
startReplication +
+ " AND replication_type = '" + startReplicationType + "'");
+
+ if (startMigration)
+ {
+ schemaChange(String.format("CREATE TABLE %s.tbl (pk int PRIMARY
KEY, val int)", ksName));
+ schemaChange(String.format("ALTER KEYSPACE %s WITH
replication_type = 'tracked'", ksName));
+
assertThat(ClusterMetadata.current().mutationTrackingMigrationState.isMigrating(ksName)).isTrue();
+ }
+
+ Map<String, String> before = replicationOf(ksName);
+ String proposed = proposedReplication.startsWith("{")
+ ? proposedReplication
+ : "{'class': 'SimpleStrategy', 'replication_factor':
'" + proposedReplication + "'}";
+ String alter = "ALTER KEYSPACE " + ksName + " WITH replication = " +
proposed +
+ (proposedReplicationType == null ? "" : " AND
replication_type = '" + proposedReplicationType + "'");
+
+ if (expectedRejection != null)
+ {
+ assertThatThrownBy(() -> schemaChange(alter))
+ .hasRootCauseInstanceOf(ConfigurationException.class)
+ .hasRootCauseMessage(expectedRejection.message(ksName));
+ assertThat(replicationOf(ksName)).isEqualTo(before);
+
assertThat(trackedOf(ksName)).isEqualTo(TRACKED.equals(startReplicationType) ||
startMigration);
+ }
+ else
+ {
+ schemaChange(alter);
+ assertThat(replicationOf(ksName)).isEqualTo(optionsOf(proposed));
+ if (proposedReplicationType != null)
+
assertThat(trackedOf(ksName)).isEqualTo("tracked".equals(proposedReplicationType));
+ }
+ }
+
+ /**
+ * The options in a CQL replication map, in the form {@link
ReplicationParams#asMap} returns them:
+ * the strategy class is fully qualified there.
+ */
+ private static Map<String, String> optionsOf(String cqlReplicationMap)
+ {
+ Map<String, String> options = new HashMap<>();
+ for (String entry : cqlReplicationMap.replaceAll("[{}']",
"").split(","))
+ {
+ String[] keyAndValue = entry.split(":");
+ String key = keyAndValue[0].trim();
+ String value = keyAndValue[1].trim();
+ options.put(key, ReplicationParams.CLASS.equals(key) ?
"org.apache.cassandra.locator." + value : value);
+ }
+ return options;
+ }
+
+ private static Map<String, String> replicationOf(String ksName)
+ {
+ return keyspace(ksName).params.replication.asMap();
+ }
+
+ private static boolean trackedOf(String ksName)
+ {
+ return keyspace(ksName).params.replicationType.isTracked();
+ }
+
+ private static KeyspaceMetadata keyspace(String ksName)
+ {
+ return ClusterMetadata.current().schema.getKeyspaceMetadata(ksName);
+ }
+}
diff --git
a/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java
b/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java
index 81b8621e99..4192cbffa7 100644
---
a/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java
+++
b/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java
@@ -36,12 +36,15 @@ import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.replication.MutationJournal;
import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
import
org.apache.cassandra.service.replication.migration.KeyspaceMigrationInfo;
import
org.apache.cassandra.service.replication.migration.MutationTrackingMigrationState;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
import static org.apache.cassandra.cql3.CQLTester.schemaChange;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -375,6 +378,106 @@ public class AlterSchemaMutationTrackingTest
}
}
+ @Test
+ public void testReadRepairAllowedOnWitnessKeyspace()
+ {
+ String ksName = nextKsName();
+ schemaChange("CREATE KEYSPACE " + ksName +
+ " WITH replication = {'class': 'SimpleStrategy',
'replication_factor': '3/1'}" +
+ " AND replication_type = 'tracked'");
+
+ // CREATE TABLE with a repairing strategy
+ schemaChange(String.format("CREATE TABLE %s.created (pk int PRIMARY
KEY, val int)" +
+ " WITH read_repair = 'BLOCKING'", ksName));
+ assertEquals(ReadRepairStrategy.BLOCKING, readRepairOf(ksName,
"created"));
+
+ // ALTER TABLE from NONE to a repairing strategy
+ schemaChange(String.format("CREATE TABLE %s.altered (pk int PRIMARY
KEY, val int)" +
+ " WITH read_repair = 'NONE'", ksName));
+ schemaChange(String.format("ALTER TABLE %s.altered WITH read_repair =
'BLOCKING'", ksName));
+ assertEquals(ReadRepairStrategy.BLOCKING, readRepairOf(ksName,
"altered"));
+
+ // CREATE TABLE LIKE a source whose strategy is repairing
+ schemaChange(String.format("CREATE TABLE %s.copied LIKE %s.created",
ksName, ksName));
+ assertEquals(ReadRepairStrategy.BLOCKING, readRepairOf(ksName,
"copied"));
+ }
+
+ private static ReadRepairStrategy readRepairOf(String keyspace, String
table)
+ {
+ return ClusterMetadata.current()
+ .schema
+ .getKeyspaceMetadata(keyspace)
+ .getTableOrViewNullable(table)
+ .params
+ .readRepair;
+ }
+
+ @Test
+ public void testSecondaryIndexesAllowedWithWitnesses()
+ {
+ // index first, then witnesses
+ String indexFirstKs = nextKsName();
+ schemaChange("CREATE KEYSPACE " + indexFirstKs +
+ " WITH replication = {'class': 'SimpleStrategy',
'replication_factor': '3'}" +
+ " AND replication_type = 'tracked'");
+ schemaChange(String.format("CREATE TABLE %s.tbl (pk int PRIMARY KEY,
val int)", indexFirstKs));
+ schemaChange(String.format("CREATE INDEX ON %s.tbl (val)",
indexFirstKs));
+
+ schemaChange("ALTER KEYSPACE " + indexFirstKs +
+ " WITH replication = {'class': 'SimpleStrategy',
'replication_factor': '3/1'}");
+
+
assertTrue(ClusterMetadata.current().schema.getKeyspaceMetadata(indexFirstKs)
+
.replicationStrategy.getReplicationFactor().hasTransientReplicas());
+
+ // witnesses first, then index
+ String witnessFirstKs = nextKsName();
+ schemaChange("CREATE KEYSPACE " + witnessFirstKs +
+ " WITH replication = {'class': 'SimpleStrategy',
'replication_factor': '3/1'}" +
+ " AND replication_type = 'tracked'");
+ schemaChange(String.format("CREATE TABLE %s.tbl (pk int PRIMARY KEY,
val int)", witnessFirstKs));
+ schemaChange(String.format("CREATE INDEX ON %s.tbl (val)",
witnessFirstKs));
+
+
assertFalse(ClusterMetadata.current().schema.getKeyspaceMetadata(witnessFirstKs)
+
.getTableOrViewNullable("tbl").indexes.isEmpty());
+ }
+
+ @Test
+ public void testWitnessPromotionSkippableAtRuntime()
+ {
+ String ksName = nextKsName();
+ schemaChange("CREATE KEYSPACE " + ksName +
+ " WITH replication = {'class': 'SimpleStrategy',
'replication_factor': '3/1'}" +
+ " AND replication_type = 'tracked'");
+ String promote = "ALTER KEYSPACE " + ksName +
+ " WITH replication = {'class': 'SimpleStrategy',
'replication_factor': '3'}";
+
+ assertThatThrownBy(() -> schemaChange(promote))
+ .hasRootCauseInstanceOf(ConfigurationException.class);
+
+ StorageService.instance.setAllowUnsafeWitnessPromotion(true);
+ try
+ {
+
assertTrue(StorageService.instance.getAllowUnsafeWitnessPromotion());
+ schemaChange(promote);
+
assertFalse(ClusterMetadata.current().schema.getKeyspaceMetadata(ksName)
+
.replicationStrategy.getReplicationFactor().hasTransientReplicas());
+ }
+ finally
+ {
+ StorageService.instance.setAllowUnsafeWitnessPromotion(false);
+ }
+
+ // Cleared again, so a second keyspace is still protected
+ String stillGuarded = nextKsName();
+ schemaChange("CREATE KEYSPACE " + stillGuarded +
+ " WITH replication = {'class': 'SimpleStrategy',
'replication_factor': '3/1'}" +
+ " AND replication_type = 'tracked'");
+ assertThatThrownBy(() ->
+ schemaChange("ALTER KEYSPACE " + stillGuarded +
+ " WITH replication = {'class': 'SimpleStrategy',
'replication_factor': '3'}"))
+ .hasRootCauseInstanceOf(ConfigurationException.class);
+ }
+
/**
* The mutation journal can't be disabled for tracked replication, so
attempting to set durable_writes=false
* on tracked keyspaces needs to fail validation
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
