diff --git a/doc/modules/cassandra/pages/architecture/dynamo.adoc b/doc/modules/cassandra/pages/architecture/dynamo.adoc index 551d5a33388c..5da99becc39f 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 `/ full lacks the pending state @@ -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 @@ -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 before = replicationFactorsByGroup(current); + Map 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 replicationFactorsByGroup(AbstractReplicationStrategy strategy) + { + if (!(strategy instanceof NetworkTopologyStrategy)) + return Collections.singletonMap("", strategy.getReplicationFactor()); + + NetworkTopologyStrategy nts = (NetworkTopologyStrategy) strategy; + Map 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() { 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 cebc174e2418..c544ad867e4c 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.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 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); 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 1250cb79ec6b..c92dfed4b386 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.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 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); 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 fc1c2343f190..31e9b920a1db 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 99eeef4a588f..7eca3ab992d4 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 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); diff --git a/src/java/org/apache/cassandra/locator/ReplicaPlans.java b/src/java/org/apache/cassandra/locator/ReplicaPlans.java index 01de6d4a5d62..34bca5d18a7d 100644 --- a/src/java/org/apache/cassandra/locator/ReplicaPlans.java +++ b/src/java/org/apache/cassandra/locator/ReplicaPlans.java @@ -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())); diff --git a/src/java/org/apache/cassandra/replication/TrackedWriteRequest.java b/src/java/org/apache/cassandra/replication/TrackedWriteRequest.java index 9a5ab64157fc..1b0bb5856d8d 100644 --- a/src/java/org/apache/cassandra/replication/TrackedWriteRequest.java +++ b/src/java/org/apache/cassandra/replication/TrackedWriteRequest.java @@ -195,7 +195,14 @@ public static AbstractWriteResponseHandler perform( 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 15512ac98107..5bfd2d1fa22f 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.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,17 @@ public boolean getSkipPaxosRepairCompatibilityCheck() return PaxosRepair.getSkipPaxosRepairCompatibilityCheck(); } + public void setAllowUnsafeWitnessPromotion(boolean allow) + { + AlterKeyspaceStatement.setAllowUnsafeWitnessPromotion(allow); + logger.info("AllowUnsafeWitnessPromotion set to {} via jmx", allow); + } + + 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 1e6b8e6a46ea..28e44de9a71e 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -1366,6 +1366,9 @@ public void enableAuditLog(String loggerName, String includedKeyspaces, String e 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 97de0841e3f5..4367d2ed8a51 100644 --- a/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java +++ b/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java @@ -31,6 +31,8 @@ import javax.annotation.Nonnull; import javax.annotation.Nullable; +import com.google.common.annotations.VisibleForTesting; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -447,24 +449,10 @@ private static > void startTracked(PaxosPrepare pre Message selfMessage = null; Message 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(); @@ -506,6 +494,23 @@ else if (replica == dataNode) } } + @VisibleForTesting + static Replica selectDataNode(Participants participants, InetAddressAndPort local) + { + Replica localReplica = participants.lookup(local); + if (localReplica != null && localReplica.isFull()) + 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 > void startUntracked(PaxosPrepare prepare, Participants participants, Message send, BiFunction> 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 656b3be4f594..611491b93948 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 @@ * * 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 000000000000..6c9fef22a23b --- /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 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 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 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 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 000000000000..97e1da1ee00d --- /dev/null +++ b/test/unit/org/apache/cassandra/service/paxos/PaxosWitnessDataNodeTest.java @@ -0,0 +1,219 @@ +/* + * 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.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 isAlive) + { + return participants(KEYSPACE, isAlive); + } + + private static Paxos.Participants participants(String keyspace, Predicate isAlive) + { + TableMetadata table = Schema.instance.getTableMetadata(keyspace, TABLE); + assertNotNull("test schema was not created", table); + Token token = table.partitioner.getToken(ByteBufferUtil.bytes(1)); + 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 fullReplicas(Paxos.Participants participants) + { + List 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 witnesses(Paxos.Participants participants) + { + List 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()); + } + + /** + * 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 000000000000..f33c8b37ade8 --- /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 000000000000..7d7c980930fd --- /dev/null +++ b/test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceWitnessTransitionTest.java @@ -0,0 +1,282 @@ +/* + * 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 transitions() + { + List 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")); + + // 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 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 optionsOf(String cqlReplicationMap) + { + Map 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 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 81b8621e9902..8fe2bac73a90 100644 --- a/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java +++ b/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java @@ -34,14 +34,18 @@ import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.exceptions.ConfigurationException; +import org.apache.cassandra.locator.ReplicationFactor; 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 +379,134 @@ private void assertStatesEqual(MutationTrackingMigrationState expected, Mutation } } + @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; + } + + /** + * Adding a full replica before dropping the witness keeps the replica count while the data is + * redistributed. The full repair the client warning asks for after the first step is what + * populates the promoted replica. + */ + @Test + public void testRemoveWitnessesByAddingAFullReplicaFirst() + { + String ksName = nextKsName(); + schemaChange("CREATE KEYSPACE " + ksName + + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '3/1'}" + + " AND replication_type = 'tracked'"); + + schemaChange("ALTER KEYSPACE " + ksName + + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '4/1'}"); + ReplicationFactor intermediate = ClusterMetadata.current().schema.getKeyspaceMetadata(ksName) + .replicationStrategy.getReplicationFactor(); + assertEquals(3, intermediate.fullReplicas); + assertEquals(1, intermediate.transientReplicas()); + + schemaChange("ALTER KEYSPACE " + ksName + + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '3'}"); + ReplicationFactor finalRf = ClusterMetadata.current().schema.getKeyspaceMetadata(ksName) + .replicationStrategy.getReplicationFactor(); + assertEquals(3, finalRf.fullReplicas); + assertFalse(finalRf.hasTransientReplicas()); + } + + @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