diff --git a/CHANGES.txt b/CHANGES.txt index dbeb6c80f1..83b3bb7d67 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -2,6 +2,7 @@ * Fix a removed TTLed row re-appearance in a materialized view after a cursor compaction (CASSANDRA-21152) * Rework ZSTD dictionary compression logic to create a trainer per training (CASSANDRA-21209) Merged from 5.0: + * Backport Automated Repair Inside Cassandra for CEP-37 (CASSANDRA-21138) Merged from 4.1: Merged from 4.0: * Rate limit password changes (CASSANDRA-21202) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 42935df4d6..fdfb98403d 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -2761,11 +2761,7 @@ storage_compatibility_mode: NONE # repair as anti-compaction during repair may contribute to additional space temporarily. # if you want to disable this feature (the recommendation is not to, but if you want to disable it for whatever reason) # then set the ratio to 0.0 -# repair_disk_headroom_reject_ratio: 0.2; - -# This is the deprecated config which was used to safeguard incremental repairs. Use repair_disk_headroom_reject_ratio -# instead as it safeguards against all repairs. -# incremental_repair_disk_headroom_reject_ratio: 0.2; +# repair_disk_headroom_reject_ratio: 0.2 # Configuration for Auto Repair Scheduler. # diff --git a/conf/cassandra_latest.yaml b/conf/cassandra_latest.yaml index 13b54aa53f..382bef78d9 100644 --- a/conf/cassandra_latest.yaml +++ b/conf/cassandra_latest.yaml @@ -2506,7 +2506,7 @@ storage_compatibility_mode: NONE # repair as anti-compaction during repair may contribute to additional space temporarily. # if you want to disable this feature (the recommendation is not to, but if you want to disable it for whatever reason) # then set the ratio to 0.0 -# repair_disk_headroom_reject_ratio: 0.2; +# repair_disk_headroom_reject_ratio: 0.2 # Configuration for Auto Repair Scheduler. # diff --git a/src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java b/src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java index 967fe540eb..f0e31c4e59 100644 --- a/src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java +++ b/src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java @@ -132,6 +132,7 @@ public class AutoRepair } AutoRepairUtils.setup(); + AutoRepairUtils.migrateAutoRepairHistoryForUpgrade(); for (AutoRepairConfig.RepairType repairType : AutoRepairConfig.RepairType.values()) { diff --git a/src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java b/src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java index d1a10f9d7e..e25da51157 100644 --- a/src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java +++ b/src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java @@ -237,6 +237,76 @@ public class AutoRepairUtils ConsistencyLevel.LOCAL_QUORUM : ConsistencyLevel.ONE; } + /** + * Migrates auto_repair_history and auto_repair_priority entries from the pre-upgrade + * host ID to the post-upgrade host ID (NodeId-derived UUID). + * No-op if the node was not upgraded or migration already happened. + * Called once during AutoRepair.setup(), before repair scheduling begins. + */ + public static void migrateAutoRepairHistoryForUpgrade() + { + try + { + Directory directory = ClusterMetadata.current().directory; + NodeId myNodeId = directory.peerId(FBUtilities.getBroadcastAddressAndPort()); + if (myNodeId == null) + return; + + UUID oldHostId = directory.hostId(myNodeId); + UUID newHostId = myNodeId.toUUID(); + + if (oldHostId.equals(newHostId)) + { + logger.debug("No host ID migration needed — old and new IDs are identical ({})", newHostId); + return; + } + + logger.info("Migrating auto-repair history from pre-upgrade host ID {} to new host ID {}", oldHostId, newHostId); + + for (RepairType repairType : RepairType.values()) + { + // Migrate auto_repair_history using the same distributed read/write path as AutoRepair + List histories = getAutoRepairHistory(repairType); + if (histories != null) + { + for (AutoRepairHistory entry : histories) + { + if (entry.hostId.equals(oldHostId)) + { + // Insert new entry with the post-upgrade host ID, preserving timestamps + insertNewRepairHistory(repairType, newHostId, entry.lastRepairStartTime, entry.lastRepairFinishTime); + // Update start timestamp and repair turn to match the original entry + if (entry.repairTurn != null) + updateStartAutoRepairHistory(repairType, newHostId, entry.lastRepairStartTime, RepairTurn.valueOf(entry.repairTurn)); + // Delete the old entry + deleteAutoRepairHistory(repairType, oldHostId); + logger.info("Migrated auto_repair_history for repair type {} from {} to {}", repairType, oldHostId, newHostId); + break; + } + } + } + + // Migrate auto_repair_priority + Set priorityIds = getPriorityHostIds(repairType); + if (priorityIds.contains(oldHostId)) + { + removePriorityStatus(repairType, oldHostId); + SetSerializer serializer = SetSerializer.getInstance(UUIDSerializer.instance, UTF8Type.instance.comparatorSet); + addPriorityHost.execute(QueryState.forInternalCalls(), + QueryOptions.forInternalCalls(internalQueryCL, + Lists.newArrayList(serializer.serialize(Collections.singleton(newHostId)), + ByteBufferUtil.bytes(repairType.toString()))), + Dispatcher.RequestTime.forImmediateExecution()); + logger.info("Migrated auto_repair_priority for repair type {} from {} to {}", repairType, oldHostId, newHostId); + } + } + } + catch (Exception e) + { + logger.error("Failed to migrate auto-repair history for upgrade", e); + } + } + public static class AutoRepairHistory { UUID hostId; diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java index 791465eb60..6f18007c67 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@ -125,6 +125,7 @@ import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.net.NoPayload; import org.apache.cassandra.net.Verb; +import org.apache.cassandra.repair.autorepair.AutoRepair; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.service.ActiveRepairService; @@ -926,6 +927,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance ActiveRepairService.instance().start(); StreamManager.instance.start(); PaxosState.startAutoRepairs(); + StorageService.instance.doAutoRepairSetup(); CassandraDaemon.getInstanceForTesting().completeSetup(); } @@ -1024,6 +1026,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance () -> SSTableReader.shutdownBlocking(1L, MINUTES), () -> shutdownAndWait(Collections.singletonList(ActiveRepairService.repairCommandExecutor())), () -> ActiveRepairService.instance().shutdownNowAndWait(1L, MINUTES), + () -> AutoRepair.instance.shutdownBlocking(), () -> EpochAwareDebounce.instance.close(), SnapshotManager.instance::close, () -> IndexStatusManager.instance.shutdownAndWait(1L, MINUTES), diff --git a/test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerStatsHelper.java b/test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerStatsHelper.java index e8d5ddd564..2dea0dbabe 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerStatsHelper.java +++ b/test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerStatsHelper.java @@ -119,10 +119,6 @@ public class AutoRepairSchedulerStatsHelper extends TestBaseImpl public static void testSchedulerStats() throws ParseException { - // ensure there was no history of previous repair runs through the scheduler - Object[][] rows = cluster.coordinator(1).execute(String.format("SELECT repair_type, host_id, repair_start_ts, repair_finish_ts, repair_turn FROM %s.%s", DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY), ConsistencyLevel.QUORUM); - assertEquals(0, rows.length); - // disabling AutoRepair for system_distributed and system_auth tables to avoid // interfering with the repaired bytes/plans calculation disableAutoRepair(SystemDistributedKeyspace.NAME, SystemDistributedKeyspace.TABLE_NAMES); diff --git a/test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerTest.java b/test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerTest.java index 9e111e2bb6..ac5a764d6d 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/repair/AutoRepairSchedulerTest.java @@ -42,7 +42,6 @@ import org.apache.cassandra.metrics.AutoRepairMetricsManager; import org.apache.cassandra.repair.autorepair.AutoRepair; import org.apache.cassandra.repair.autorepair.AutoRepairConfig; import org.apache.cassandra.schema.SystemDistributedKeyspace; -import org.apache.cassandra.service.AutoRepairService; import org.apache.cassandra.utils.FBUtilities; import static org.apache.cassandra.schema.SchemaConstants.DISTRIBUTED_KEYSPACE_NAME; @@ -120,22 +119,6 @@ public class AutoRepairSchedulerTest extends TestBaseImpl @Test public void testScheduler() throws ParseException { - // ensure there was no history of previous repair runs through the scheduler - Object[][] rows = cluster.coordinator(1).execute(String.format("SELECT repair_type, host_id, repair_start_ts, repair_finish_ts, repair_turn FROM %s.%s", DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY), ConsistencyLevel.QUORUM); - assertEquals(0, rows.length); - - cluster.forEach(i -> i.runOnInstance(() -> { - try - { - AutoRepairService.setup(); - AutoRepair.instance.setup(); - } - catch (Exception e) - { - throw new RuntimeException(e); - } - })); - // validate that the repair ran on all nodes cluster.forEach(i -> i.runOnInstance(() -> { String broadcastAddress = FBUtilities.getJustBroadcastAddress().toString(); diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/AutoRepairUpgradeTest.java b/test/distributed/org/apache/cassandra/distributed/upgrade/AutoRepairUpgradeTest.java new file mode 100644 index 0000000000..4986156756 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/AutoRepairUpgradeTest.java @@ -0,0 +1,248 @@ +/* + * 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.upgrade; + +import java.util.Date; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import com.google.common.collect.ImmutableMap; + +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.distributed.UpgradeableCluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.repair.autorepair.AutoRepairConfig; + +import static org.awaitility.Awaitility.await; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +/** + * Upgrade test for auto-repair verifying that it runs successfully before and after + * upgrading from 5.0 to current. The first repair round executes on 5.0 nodes + * (automatically during startup) and the second round after all nodes are upgraded. + * + * Host IDs change across the upgrade (from random UUIDs to NodeId-derived UUIDs). + * The migration in {@code AutoRepairUtils.migrateAutoRepairHistoryForUpgrade()} re-keys + * entries under the new host IDs, preserving repair timestamps. The test verifies that: + *
    + *
  1. 3 repair history entries exist before the upgrade (on 5.0)
    2. + *
  2. 3 entries exist after upgrade, keyed by new host IDs, retaining per-node pre-upgrade timestamps
    3. + *
  3. After repair runs, each entry's timestamp exceeds its own pre-upgrade value
    4. + *
+ * + * Auto-repair is started automatically during node startup via + * {@code StorageService.doAutoRepairSetup()} when the config is enabled. + * In 5.0, the JVM property {@code cassandra.autorepair.enable=true} is also required. + */ +public class AutoRepairUpgradeTest extends UpgradeTestBase +{ + private static final Logger logger = LoggerFactory.getLogger(AutoRepairUpgradeTest.class); + + @Test + public void testAutoRepairAcrossUpgrade() throws Throwable + { + // 5.0 requires this JVM property to enable auto-repair (schema tables, JMX, scheduler). + // Trunk does not use this property. + System.setProperty("cassandra.autorepair.enable", "true"); // checkstyle: suppress nearby 'blockSystemPropertyUsage' + + // Maps pre-upgrade host ID -> finish timestamp, captured right before upgrade + Map preUpgradeTimestamps = new ConcurrentHashMap<>(); + + new TestCase() + .nodes(3) + .singleUpgradeToCurrentFrom(v50) + .withConfig(cfg -> cfg.with(Feature.NETWORK, Feature.GOSSIP) + .set("auto_repair", + ImmutableMap.of( + "repair_type_overrides", + ImmutableMap.of(AutoRepairConfig.RepairType.FULL.getConfigName(), + ImmutableMap.of( + "initial_scheduler_delay", "60s", + "enabled", "true", + "parallel_repair_count", "3", + "allow_parallel_replica_repair", "true", + "min_repair_interval", "60s")))) + .set("auto_repair.enabled", "true") + .set("auto_repair.global_settings.repair_by_keyspace", "true") + .set("auto_repair.global_settings.repair_retry_backoff", "5s") + .set("auto_repair.repair_task_min_duration", "0s") + .set("auto_repair.repair_check_interval", "60s")) + .setup(cluster -> { + cluster.schemaChange("CREATE KEYSPACE IF NOT EXISTS " + KEYSPACE + + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3};"); + cluster.schemaChange("CREATE TABLE IF NOT EXISTS " + KEYSPACE + + ".tbl (pk int, ck text, v1 int, v2 int, PRIMARY KEY (pk, ck))"); + + // Wait for auto-repair to complete on all 5.0 nodes. + waitForNEntries(cluster, 3); + + assertEquals("Expected repair history for all 3 nodes on 5.0", + 3, captureFinishTimestamps(cluster).size()); + }) + .runBeforeClusterUpgrade(cluster -> { + // Wait for any in-flight repair to complete before capturing timestamps. + waitForNoInFlightRepairs(cluster); + preUpgradeTimestamps.putAll(captureFinishTimestamps(cluster)); + logger.info("Pre-upgrade timestamps: {}", preUpgradeTimestamps); + + // Seed auto_repair_priority with pre-upgrade host IDs to test priority migration. + String hostIdSet = preUpgradeTimestamps.keySet().stream() + .map(id -> id.toString()) + .collect(Collectors.joining(", ")); + cluster.coordinator(1).execute( + String.format("INSERT INTO system_distributed.auto_repair_priority (repair_type, repair_priority) VALUES ('%s', {%s})", + AutoRepairConfig.RepairType.FULL.toString(), hostIdSet), + ConsistencyLevel.QUORUM); + logger.info("Seeded auto_repair_priority with pre-upgrade host IDs: {}", preUpgradeTimestamps.keySet()); + }) + .runAfterClusterUpgrade(cluster -> { + // Phase 1: Verify migration — old entries replaced by new entries with + // different host IDs but preserved per-node timestamps + Map migratedTimestamps = captureFinishTimestamps(cluster); + logger.info("Pre-upgrade entries: {}, post-migration entries: {}", preUpgradeTimestamps, migratedTimestamps); + assertEquals("Expected exactly 3 migrated entries", 3, migratedTimestamps.size()); + + // Host IDs must have changed — new entries should not use pre-upgrade IDs + for (String id : migratedTimestamps.keySet()) + assertFalse("Migrated entry should use new host ID, not pre-upgrade ID " + id, + preUpgradeTimestamps.containsKey(id)); + + // Each migrated entry should retain its exact original per-node timestamp. + // Since host IDs changed, we compare by value: every migrated timestamp must + // exist in the pre-upgrade set (values preserved exactly during migration). + for (Long ts : migratedTimestamps.values()) + assertTrue("Migrated timestamp " + ts + " should match a pre-upgrade timestamp", + preUpgradeTimestamps.containsValue(ts)); + + // Verify auto_repair_priority migration: old host IDs should be replaced by new ones, + // and the total count should remain the same (3 entries seeded before upgrade). + Set priorityIds = capturePriorityHostIds(cluster); + logger.info("Post-migration priority IDs: {}", priorityIds); + assertEquals("Priority set should have same number of entries after migration", + preUpgradeTimestamps.size(), priorityIds.size()); + for (String id : priorityIds) + assertFalse("Priority should not contain pre-upgrade host ID " + id, + preUpgradeTimestamps.containsKey(id)); + + // Phase 2: Wait for repair to run (after initial_scheduler_delay expires), + // then verify each entry's timestamp exceeds its own migrated value. + Map migratedSnapshot = new HashMap<>(migratedTimestamps); + waitForAllTimestampsExceeded(cluster, migratedSnapshot); + + Map postRepairTimestamps = captureFinishTimestamps(cluster); + assertEquals("Expected 3 entries after repair", 3, postRepairTimestamps.size()); + assertEquals("Post-repair entries should use same host IDs as migrated", + migratedSnapshot.keySet(), postRepairTimestamps.keySet()); + for (Map.Entry entry : postRepairTimestamps.entrySet()) + assertTrue("Post-repair timestamp for " + entry.getKey() + " should exceed migrated timestamp", + entry.getValue() > migratedSnapshot.get(entry.getKey())); + + // Priority table should be cleared after repair completes + Set postRepairPriorityIds = capturePriorityHostIds(cluster); + assertTrue("Priority set should be empty after post-upgrade repair completes, but was: " + postRepairPriorityIds, + postRepairPriorityIds.isEmpty()); + }) + .run(); + } + + private void waitForNEntries(UpgradeableCluster cluster, int expected) + { + await().atMost(5, TimeUnit.MINUTES) + .pollInterval(2, TimeUnit.SECONDS) + .until(() -> captureFinishTimestamps(cluster).size() >= expected); + } + + private void waitForAllTimestampsExceeded(UpgradeableCluster cluster, Map baseline) + { + await().atMost(5, TimeUnit.MINUTES) + .pollInterval(2, TimeUnit.SECONDS) + .until(() -> { + Map current = captureFinishTimestamps(cluster); + if (current.size() < baseline.size()) + return false; + for (Map.Entry entry : baseline.entrySet()) + { + Long currentTs = current.get(entry.getKey()); + if (currentTs == null || currentTs <= entry.getValue()) + return false; + } + return true; + }); + } + + private Set capturePriorityHostIds(UpgradeableCluster cluster) + { + Object[][] rows = cluster.coordinator(1).execute( + String.format("SELECT repair_priority FROM system_distributed.auto_repair_priority WHERE repair_type='%s'", + AutoRepairConfig.RepairType.FULL.toString()), + ConsistencyLevel.QUORUM); + if (rows.length == 0 || rows[0][0] == null) + return Set.of(); + @SuppressWarnings("unchecked") + Set uuids = (Set) rows[0][0]; + return uuids.stream().map(UUID::toString).collect(Collectors.toSet()); + } + + private void waitForNoInFlightRepairs(UpgradeableCluster cluster) + { + await().atMost(2, TimeUnit.MINUTES) + .pollInterval(1, TimeUnit.SECONDS) + .until(() -> { + Object[][] rows = cluster.coordinator(1).execute( + String.format("SELECT host_id, repair_start_ts, repair_finish_ts FROM system_distributed.auto_repair_history WHERE repair_type='%s'", + AutoRepairConfig.RepairType.FULL.toString()), + ConsistencyLevel.QUORUM); + for (Object[] row : rows) + { + long startTs = ((Date) row[1]).getTime(); + long finishTs = ((Date) row[2]).getTime(); + if (startTs > finishTs) + return false; // repair still in flight + } + return true; + }); + } + + private Map captureFinishTimestamps(UpgradeableCluster cluster) + { + Object[][] rows = cluster.coordinator(1).execute( + String.format("SELECT host_id, repair_finish_ts FROM system_distributed.auto_repair_history WHERE repair_type='%s'", + AutoRepairConfig.RepairType.FULL.toString()), + ConsistencyLevel.QUORUM); + Map timestamps = new HashMap<>(); + for (Object[] row : rows) + { + String hostId = row[0].toString(); + long finishTs = ((Date) row[1]).getTime(); + timestamps.put(hostId, finishTs); + } + return timestamps; + } +}