From b6aa6c3bdb790bff11069d9e6ad936ff6d52a2cf Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Tue, 8 Sep 2026 15:33:12 -0700 Subject: [PATCH 1/7] Allow any read_repair value on a witness keyspace Witness replicas require mutation tracking, and tracked reads never build a repairing ReadRepair, so the read_repair table option cannot affect a read that reaches a witness. CASSANDRA-20930 disabled this check in CreateTableStatement but left the AlterTableStatement and CopyTableStatement copies in place. --- .../schema/AlterTableStatement.java | 7 ---- .../statements/schema/CopyTableStatement.java | 7 ---- .../schema/CreateTableStatement.java | 7 ---- .../AlterSchemaMutationTrackingTest.java | 35 +++++++++++++++++++ 4 files changed, 35 insertions(+), 21 deletions(-) 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/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/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java b/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java index 81b8621e9902..145f561f564c 100644 --- a/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java +++ b/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java @@ -38,6 +38,7 @@ import org.apache.cassandra.schema.TableId; import org.apache.cassandra.service.replication.migration.KeyspaceMigrationInfo; import org.apache.cassandra.service.replication.migration.MutationTrackingMigrationState; +import org.apache.cassandra.service.reads.repair.ReadRepairStrategy; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.Epoch; @@ -375,6 +376,40 @@ 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; + } + /** * The mutation journal can't be disabled for tracked replication, so attempting to set durable_writes=false * on tracked keyspaces needs to fail validation From d51c22529d4bd7503f1edfc58f2792547a0f9c04 Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Thu, 10 Sep 2026 09:04:27 -0700 Subject: [PATCH 2/7] Reject adding witnesses while a mutation tracking migration is in flight Reads for a range still pending migration take the untracked path, which refuses transient replicas, and those ranges rely on blocking read repair that a witness cannot serve. Two conditions are needed because AlterSchema starts the migration after the statement validates, so a statement that both enables tracking and adds witnesses cannot be seen in the migration state. --- .../schema/AlterKeyspaceStatement.java | 33 +++ ...ceStatementUnsafeTransientChangesTest.java | 98 +++++++++ .../AlterKeyspaceWitnessTransitionTest.java | 206 ++++++++++++++++++ .../AlterSchemaMutationTrackingTest.java | 1 + 4 files changed, 338 insertions(+) create mode 100644 test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceStatementUnsafeTransientChangesTest.java create mode 100644 test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceWitnessTransitionTest.java 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 30faee655e27..eb6f10c15a73 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java @@ -104,6 +104,7 @@ public Keyspaces apply(ClusterMetadata metadata) validateNoRangeMovements(); validateTransientReplication(keyspace, newKeyspace); + validateWitnessesNotAddedDuringMigration(metadata, keyspace, newKeyspace); // Because we used to not properly validate unrecognized options, we only log a warning if we find one. try @@ -247,6 +248,38 @@ 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 keyspace migrating from untracked to tracked replication has pending ranges, and reads for a + * pending range take the untracked path (see MigrationRouter#shouldUseTrackedForReads). Those + * reads would contact a transient replica, which the untracked path rejects outright in + * RangeCommandIterator#executeNormal. Migration of a pending range also relies on blocking read + * repair to converge the replicas, and a witness cannot participate in read repair. + */ + private void validateWitnessesNotAddedDuringMigration(ClusterMetadata metadata, + KeyspaceMetadata current, + KeyspaceMetadata proposed) + { + // Opt out alongside the other transient replication rules, see validateTransientReplication + if (allow_unsafe_transient_changes) + return; + + if (!proposed.replicationStrategy.getReplicationFactor().hasTransientReplicas()) + return; + + if (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 (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/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..cc17bf6f1440 --- /dev/null +++ b/test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceStatementUnsafeTransientChangesTest.java @@ -0,0 +1,98 @@ +/* + * 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.locator.ReplicationFactor; +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.assertEquals; +import static org.junit.Assert.assertFalse; +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..3ef3572ff680 --- /dev/null +++ b/test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceWitnessTransitionTest.java @@ -0,0 +1,206 @@ +/* + * 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.ArrayList; +import java.util.Collection; +import java.util.List; +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.exceptions.ConfigurationException; +import org.apache.cassandra.locator.ReplicationFactor; +import org.apache.cassandra.replication.MutationJournal; +import org.apache.cassandra.schema.KeyspaceMetadata; +import org.apache.cassandra.tcm.ClusterMetadata; + +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 factor and replication type transitions a keyspace with witnesses (transient + * replicas) permits. Each rejected transition names the rule that rejects it, because several rules + * overlap and a test asserting only that the statement failed can pass for the wrong reason. + */ +@RunWith(Parameterized.class) +public class AlterKeyspaceWitnessTransitionTest +{ + private static final AtomicInteger ksCounter = new AtomicInteger(); + + /** The state a keyspace starts in, including whether a mutation tracking migration is in flight. */ + enum Start + { + TRACKED_3("3", "tracked", false), + TRACKED_3_1("3/1", "tracked", false), + UNTRACKED_3("3", "untracked", false), + MIGRATING_3("3", "untracked", true); + + final String replicationFactor; + final String replicationType; + final boolean startMigration; + + Start(String replicationFactor, String replicationType, boolean startMigration) + { + this.replicationFactor = replicationFactor; + this.replicationType = replicationType; + this.startMigration = startMigration; + } + } + + @Parameterized.Parameter(0) + public Start start; + + @Parameterized.Parameter(1) + public String proposedReplicationFactor; + + @Parameterized.Parameter(2) + public String proposedReplicationType; + + /** The rejection this transition must produce, or null when it is permitted. */ + @Parameterized.Parameter(3) + public String expectedRejection; + + @Parameterized.Parameter(4) + public String description; + + @Parameterized.Parameters(name = "{4}") + public static Collection transitions() + { + List rows = new ArrayList<>(); + + // A settled tracked keyspace accepts witnesses: the guard is on the migration, not on transient + // replicas as such + rows.add(row(Start.TRACKED_3, "3/1", null, null, "settled tracked keyspace accepts witnesses")); + + // Enabling tracking and adding witnesses at once starts a migration, so the migration state + // cannot yet show it and the transition itself has to be recognised + rows.add(row(Start.UNTRACKED_3, "3/1", "tracked", "Cannot enable mutation tracking on", + "enabling tracking and adding witnesses at once")); + + // Reads for a pending range take the untracked path and would contact a transient replica + rows.add(row(Start.MIGRATING_3, "3/1", null, "Cannot add transient replicas to", + "adding witnesses while a migration is in flight")); + + return rows; + } + + private static Object[] row(Start start, String rf, String type, String rejection, String description) + { + return new Object[]{ start, rf, type, rejection, description }; + } + + @BeforeClass + public static void setUpClass() throws Exception + { + CassandraRelevantProperties.PARTITIONER.setString(Murmur3Partitioner.class.getName()); + ServerTestUtils.daemonInitialization(); + ServerTestUtils.prepareServer(); + MutationJournal.start(); + } + + @Test + public void testTransition() + { + String ksName = "ks" + ksCounter.incrementAndGet(); + schemaChange("CREATE KEYSPACE " + ksName + + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '" + start.replicationFactor + "'}" + + " AND replication_type = '" + start.replicationType + "'"); + + if (start.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(); + } + + String alter = "ALTER KEYSPACE " + ksName + + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '" + proposedReplicationFactor + "'}" + + (proposedReplicationType == null ? "" : " AND replication_type = '" + proposedReplicationType + "'"); + + if (expectedRejection != null) + { + assertThatThrownBy(() -> schemaChange(alter)) + .hasRootCauseInstanceOf(ConfigurationException.class) + .hasRootCauseMessage(rejectionMessage(ksName)); + assertUnchanged(ksName); + } + else + { + schemaChange(alter); + assertProposedStateApplied(ksName); + } + } + + /** + * The rules name the keyspace, so the expected message is completed here rather than in the table. + * Matching the whole message keeps overlapping rules distinguishable: several of these transitions + * are rejected by more than one rule in principle, and only one of them should fire. + */ + private String rejectionMessage(String ksName) + { + switch (expectedRejection) + { + case "Cannot enable mutation tracking on": + return 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.", ksName); + case "Cannot add transient replicas to": + return 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.", ksName); + default: + throw new AssertionError("unhandled rejection: " + expectedRejection); + } + } + + private void assertUnchanged(String ksName) + { + KeyspaceMetadata ksm = ClusterMetadata.current().schema.getKeyspaceMetadata(ksName); + ReplicationFactor rf = ksm.replicationStrategy.getReplicationFactor(); + assertThat(rf.toString()).isEqualTo(expectedReplicationFactor(start.replicationFactor).toString()); + assertThat(ksm.params.replicationType.isTracked()).isEqualTo(startedTracked()); + } + + private void assertProposedStateApplied(String ksName) + { + KeyspaceMetadata ksm = ClusterMetadata.current().schema.getKeyspaceMetadata(ksName); + ReplicationFactor rf = ksm.replicationStrategy.getReplicationFactor(); + assertThat(rf.toString()).isEqualTo(expectedReplicationFactor(proposedReplicationFactor).toString()); + if (proposedReplicationType != null) + assertThat(ksm.params.replicationType.isTracked()).isEqualTo("tracked".equals(proposedReplicationType)); + } + + private boolean startedTracked() + { + return "tracked".equals(start.replicationType) || start.startMigration; + } + + private static ReplicationFactor expectedReplicationFactor(String spec) + { + return ReplicationFactor.fromString(spec); + } +} diff --git a/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java b/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java index 145f561f564c..8802294e11f9 100644 --- a/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java +++ b/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java @@ -34,6 +34,7 @@ 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.replication.migration.KeyspaceMigrationInfo; From 71ebf4392fe8177f9f242fda88801a89e596aaf0 Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Thu, 10 Sep 2026 09:04:58 -0700 Subject: [PATCH 3/7] Require an opt-in to promote a witness to a full replica A promoted witness holds no data for the range it witnessed and quorum reads would count it, so promotion needs cassandra.allow_unsafe_witness_promotion, which is settable over JMX because QA performs it deliberately. The comparison is per datacenter, since NetworkTopologyStrategy sums its per-datacenter factors and aggregates hide a promotion offset by a reduction elsewhere. Dropping a witness stays legal: replica ordering removes it from the replica set rather than promoting it, and the remaining full replicas already hold the data. --- .../config/CassandraRelevantProperties.java | 1 + .../schema/AlterKeyspaceStatement.java | 114 +++++++- .../cassandra/service/StorageService.java | 12 + .../service/StorageServiceMBean.java | 3 + ...ceStatementUnsafeTransientChangesTest.java | 3 - .../AlterKeyspaceWitnessTransitionTest.java | 244 ++++++++++++------ .../AlterSchemaMutationTrackingTest.java | 69 ++++- 7 files changed, 344 insertions(+), 102 deletions(-) diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index b0631d98ace6..97dcc8d5abe2 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 eb6f10c15a73..7b247bbd93d7 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.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; @@ -53,6 +59,7 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.ALLOW_ALTER_RF_DURING_RANGE_MOVEMENT; import static org.apache.cassandra.config.CassandraRelevantProperties.ALLOW_UNSAFE_TRANSIENT_CHANGES; +import static org.apache.cassandra.config.CassandraRelevantProperties.ALLOW_UNSAFE_WITNESS_PROMOTION; import static org.apache.cassandra.replication.MutationTrackingService.DISABLED_MESSAGE; public final class AlterKeyspaceStatement extends AlterSchemaStatement @@ -104,7 +111,7 @@ public Keyspaces apply(ClusterMetadata metadata) validateNoRangeMovements(); validateTransientReplication(keyspace, newKeyspace); - validateWitnessesNotAddedDuringMigration(metadata, keyspace, newKeyspace); + validateWitnessTransitions(metadata, keyspace, newKeyspace); // Because we used to not properly validate unrecognized options, we only log a warning if we find one. try @@ -238,6 +245,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 @@ -249,30 +262,103 @@ private void validateTransientReplication(KeyspaceMetadata current, KeyspaceMeta } /** - * A keyspace migrating from untracked to tracked replication has pending ranges, and reads for a - * pending range take the untracked path (see MigrationRouter#shouldUseTrackedForReads). Those - * reads would contact a transient replica, which the untracked path rejects outright in - * RangeCommandIterator#executeNormal. Migration of a pending range also relies on blocking read - * repair to converge the replicas, and a witness cannot participate in read repair. + * 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 validateWitnessesNotAddedDuringMigration(ClusterMetadata metadata, - KeyspaceMetadata current, - KeyspaceMetadata proposed) + private void validateNoWitnessPromotion(AbstractReplicationStrategy current, AbstractReplicationStrategy proposed) { - // Opt out alongside the other transient replication rules, see validateTransientReplication - if (allow_unsafe_transient_changes) + if (allowUnsafeWitnessPromotion) return; - if (!proposed.replicationStrategy.getReplicationFactor().hasTransientReplicas()) + 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; - if (metadata.mutationTrackingMigrationState.isMigrating(keyspaceName)) + 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)); - if (proposed.params.replicationType.isTracked() && !current.params.replicationType.isTracked()) + // 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 " + 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/test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceStatementUnsafeTransientChangesTest.java b/test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceStatementUnsafeTransientChangesTest.java index cc17bf6f1440..f33c8b37ade8 100644 --- a/test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceStatementUnsafeTransientChangesTest.java +++ b/test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceStatementUnsafeTransientChangesTest.java @@ -25,14 +25,11 @@ import org.apache.cassandra.ServerTestUtils; import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.dht.Murmur3Partitioner; -import org.apache.cassandra.locator.ReplicationFactor; 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.assertEquals; -import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; /** diff --git a/test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceWitnessTransitionTest.java b/test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceWitnessTransitionTest.java index 3ef3572ff680..7d7c980930fd 100644 --- a/test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceWitnessTransitionTest.java +++ b/test/unit/org/apache/cassandra/tcm/transformations/AlterKeyspaceWitnessTransitionTest.java @@ -17,9 +17,12 @@ 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; @@ -30,86 +33,159 @@ 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.ReplicationFactor; +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 factor and replication type transitions a keyspace with witnesses (transient - * replicas) permits. Each rejected transition names the rule that rejects it, because several rules - * overlap and a test asserting only that the statement failed can pass for the wrong reason. + * 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 AtomicInteger ksCounter = new AtomicInteger(); - - /** The state a keyspace starts in, including whether a mutation tracking migration is in flight. */ - enum Start - { - TRACKED_3("3", "tracked", false), - TRACKED_3_1("3/1", "tracked", false), - UNTRACKED_3("3", "untracked", false), - MIGRATING_3("3", "untracked", true); + 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'}"; - final String replicationFactor; - final String replicationType; - final boolean startMigration; + private static final AtomicInteger ksCounter = new AtomicInteger(); - Start(String replicationFactor, String replicationType, boolean startMigration) - { - this.replicationFactor = replicationFactor; - this.replicationType = replicationType; - this.startMigration = startMigration; - } - } + 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 Start start; + public String startReplication; + /** The replication type the keyspace is created with. */ @Parameterized.Parameter(1) - public String proposedReplicationFactor; + public String startReplicationType; + /** Whether to leave a mutation tracking migration in flight before the alteration under test. */ @Parameterized.Parameter(2) - public String proposedReplicationType; + public boolean startMigration; - /** The rejection this transition must produce, or null when it is permitted. */ @Parameterized.Parameter(3) - public String expectedRejection; + 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; - @Parameterized.Parameters(name = "{4}") + 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<>(); - // A settled tracked keyspace accepts witnesses: the guard is on the migration, not on transient - // replicas as such - rows.add(row(Start.TRACKED_3, "3/1", null, null, "settled tracked keyspace accepts witnesses")); - - // Enabling tracking and adding witnesses at once starts a migration, so the migration state - // cannot yet show it and the transition itself has to be recognised - rows.add(row(Start.UNTRACKED_3, "3/1", "tracked", "Cannot enable mutation tracking on", + // 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")); - // Reads for a pending range take the untracked path and would contact a transient replica - rows.add(row(Start.MIGRATING_3, "3/1", null, "Cannot add transient replicas to", - "adding witnesses while a migration is in flight")); + // 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(Start start, String rf, String type, String rejection, String description) + private static Object[] row(String startReplication, String startType, boolean startMigration, + String proposedReplication, String proposedType, Rejection rejection, + String description) { - return new Object[]{ start, rf, type, rejection, description }; + return new Object[]{ startReplication, startType, startMigration, proposedReplication, proposedType, + rejection, description }; } @BeforeClass @@ -119,88 +195,88 @@ public static void setUpClass() throws Exception 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 = {'class': 'SimpleStrategy', 'replication_factor': '" + start.replicationFactor + "'}" + - " AND replication_type = '" + start.replicationType + "'"); + schemaChange("CREATE KEYSPACE " + ksName + " WITH replication = " + startReplication + + " AND replication_type = '" + startReplicationType + "'"); - if (start.startMigration) + 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(); } - String alter = "ALTER KEYSPACE " + ksName + - " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '" + proposedReplicationFactor + "'}" + + 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(rejectionMessage(ksName)); - assertUnchanged(ksName); + .hasRootCauseMessage(expectedRejection.message(ksName)); + assertThat(replicationOf(ksName)).isEqualTo(before); + assertThat(trackedOf(ksName)).isEqualTo(TRACKED.equals(startReplicationType) || startMigration); } else { schemaChange(alter); - assertProposedStateApplied(ksName); + assertThat(replicationOf(ksName)).isEqualTo(optionsOf(proposed)); + if (proposedReplicationType != null) + assertThat(trackedOf(ksName)).isEqualTo("tracked".equals(proposedReplicationType)); } } /** - * The rules name the keyspace, so the expected message is completed here rather than in the table. - * Matching the whole message keeps overlapping rules distinguishable: several of these transitions - * are rejected by more than one rule in principle, and only one of them should fire. + * The options in a CQL replication map, in the form {@link ReplicationParams#asMap} returns them: + * the strategy class is fully qualified there. */ - private String rejectionMessage(String ksName) + private static Map optionsOf(String cqlReplicationMap) { - switch (expectedRejection) + Map options = new HashMap<>(); + for (String entry : cqlReplicationMap.replaceAll("[{}']", "").split(",")) { - case "Cannot enable mutation tracking on": - return 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.", ksName); - case "Cannot add transient replicas to": - return 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.", ksName); - default: - throw new AssertionError("unhandled rejection: " + expectedRejection); + 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 void assertUnchanged(String ksName) - { - KeyspaceMetadata ksm = ClusterMetadata.current().schema.getKeyspaceMetadata(ksName); - ReplicationFactor rf = ksm.replicationStrategy.getReplicationFactor(); - assertThat(rf.toString()).isEqualTo(expectedReplicationFactor(start.replicationFactor).toString()); - assertThat(ksm.params.replicationType.isTracked()).isEqualTo(startedTracked()); - } - - private void assertProposedStateApplied(String ksName) + private static Map replicationOf(String ksName) { - KeyspaceMetadata ksm = ClusterMetadata.current().schema.getKeyspaceMetadata(ksName); - ReplicationFactor rf = ksm.replicationStrategy.getReplicationFactor(); - assertThat(rf.toString()).isEqualTo(expectedReplicationFactor(proposedReplicationFactor).toString()); - if (proposedReplicationType != null) - assertThat(ksm.params.replicationType.isTracked()).isEqualTo("tracked".equals(proposedReplicationType)); + return keyspace(ksName).params.replication.asMap(); } - private boolean startedTracked() + private static boolean trackedOf(String ksName) { - return "tracked".equals(start.replicationType) || start.startMigration; + return keyspace(ksName).params.replicationType.isTracked(); } - private static ReplicationFactor expectedReplicationFactor(String spec) + private static KeyspaceMetadata keyspace(String ksName) { - return ReplicationFactor.fromString(spec); + 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 8802294e11f9..103e1109dd14 100644 --- a/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java +++ b/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java @@ -37,13 +37,15 @@ 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.service.reads.repair.ReadRepairStrategy; 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; @@ -411,6 +413,71 @@ private static ReadRepairStrategy readRepairOf(String keyspace, String table) .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 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 From dae585161d086d982a11343eea37be554c546fc7 Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Tue, 8 Sep 2026 19:42:52 -0700 Subject: [PATCH 4/7] Allow secondary indexes on a witness keyspace Witnesses never serve data reads, since TrackedRead picks a full replica and summaries are built from the mutation tracking log, so an index on a witness is never consulted. The restriction was also asymmetric: CREATE INDEX on an existing witness keyspace was already accepted. Also drops a constant in CreateIndexStatement that was declared and referenced nowhere. --- .../schema/AlterKeyspaceStatement.java | 5 ---- .../schema/CreateIndexStatement.java | 1 - .../AlterSchemaMutationTrackingTest.java | 29 +++++++++++++++++++ 3 files changed, 29 insertions(+), 6 deletions(-) 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 7b247bbd93d7..b39e07860ce7 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java @@ -48,7 +48,6 @@ import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.SchemaConstants; -import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.ClientState; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.membership.NodeId; @@ -233,10 +232,6 @@ private void validateTransientReplication(KeyspaceMetadata current, KeyspaceMeta if (!current.views.isEmpty()) throw new ConfigurationException("Cannot use transient replication on keyspaces using materialized views"); - - for (TableMetadata table : current.tables) - if (!table.indexes.isEmpty()) - throw new ConfigurationException("Cannot use transient replication on keyspaces using secondary indexes"); } //This is true right now because the transition from transient -> full lacks the pending state 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/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java b/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java index 103e1109dd14..8fe2bac73a90 100644 --- a/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java +++ b/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java @@ -441,6 +441,35 @@ public void testRemoveWitnessesByAddingAFullReplicaFirst() 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() { From 6daa4855c1bb53477d1080bff79a902f37eb1ace Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Tue, 8 Sep 2026 20:51:28 -0700 Subject: [PATCH 5/7] Never lead a counter write from a witness replica A counter leader resolves the increment against its local data, which a witness does not have for the range it witnesses, so it would write a value discarding every prior increment. findCounterLeaderReplica now considers only full replicas, and TrackedWriteRequest forwards rather than leading when the local replica is a witness. --- .../cassandra/locator/ReplicaPlans.java | 4 + .../replication/TrackedWriteRequest.java | 9 +- .../test/TrackedCounterForwardingTest.java | 2 + .../TrackedCounterWitnessForwardingTest.java | 188 ++++++++++++++++++ 4 files changed, 202 insertions(+), 1 deletion(-) create mode 100644 test/distributed/org/apache/cassandra/distributed/test/TrackedCounterWitnessForwardingTest.java 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/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); + } +} From 1be6ae1c4b06d0c83cfa7eb8de15ffebdec576ae Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Wed, 9 Sep 2026 08:57:52 -0700 Subject: [PATCH 6/7] Cover the Paxos data node selection on a witness keyspace A tracked prepare sends its data request to one participant and summary requests to the rest, so a witness votes without being asked for data it does not have. Extracts the selection into PaxosPrepare#selectDataNode so the test drives production code rather than a copy of it. No behaviour change. --- .../cassandra/service/paxos/PaxosPrepare.java | 35 +-- .../paxos/PaxosWitnessDataNodeTest.java | 219 ++++++++++++++++++ 2 files changed, 239 insertions(+), 15 deletions(-) create mode 100644 test/unit/org/apache/cassandra/service/paxos/PaxosWitnessDataNodeTest.java 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/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()); + } +} From 26deaff7c656bdc8441b60ec929d2549365f1e26 Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Tue, 8 Sep 2026 17:06:34 -0700 Subject: [PATCH 7/7] Correct the transient replication documentation The list of unsupported features was inherited from 4.0 and most of it no longer holds: lightweight transactions, secondary indexes and counters all work. Records what is actually enforced, why materialized views remain rejected, and the sequences for adopting and removing witnesses. --- .../cassandra/pages/architecture/dynamo.adoc | 8 +- .../operating/transientreplication.adoc | 79 +++++++++++++++++-- 2 files changed, 73 insertions(+), 14 deletions(-) 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 `/