mirror of https://github.com/apache/cassandra
Fix AutoRepair Flaky InJvm dtest
Patch by Jaydeepkumar Chovatia; Reviewed by Andy Tolbert, Chris Lohfink for CASSANDRA-20620
This commit is contained in:
jaydeepkumar1984
Andy Tolbert
parent
ccc92767ff
commit
f2bf017e6d
|
|
@ -1,4 +1,5 @@
|
|||
5.1
|
||||
* Fix AutoRepair flaky InJvm dtest (CASSANDRA-20620)
|
||||
* Increasing default for auto_repair.sstable_upper_threshold considering large Cassandra tables & revert three lines removed from CHANGES.txt due to a merge mistake (CASSANDRA-20586)
|
||||
* Fix token restrictions with MIN_TOKEN (CASSANDRO-20557)
|
||||
* Upgrade logback version to 1.5.18 and slf4j dependencies to 2.0.17 (CASSANDRA-20429)
|
||||
|
|
|
|||
|
|
@ -196,7 +196,7 @@ public class AutoRepair
|
|||
boolean primaryRangeOnly = config.getRepairPrimaryTokenRangeOnly(repairType)
|
||||
&& turn != MY_TURN_FORCE_REPAIR;
|
||||
|
||||
long startTime = timeFunc.get();
|
||||
long startTimeInMillis = timeFunc.get();
|
||||
logger.info("My host id: {}, my turn to run repair...repair primary-ranges only? {}", myId,
|
||||
config.getRepairPrimaryTokenRangeOnly(repairType));
|
||||
AutoRepairUtils.updateStartAutoRepairHistory(repairType, myId, timeFunc.get(), turn);
|
||||
|
|
@ -242,7 +242,7 @@ public class AutoRepair
|
|||
repairKeyspace(repairType, primaryRangeOnly, repairAssignments.getKeyspaceName(), repairAssignments.getRepairAssignments(), collectedRepairStats);
|
||||
}
|
||||
|
||||
cleanupAndUpdateStats(turn, repairType, repairState, myId, startTime, collectedRepairStats);
|
||||
cleanupAndUpdateStats(turn, repairType, repairState, myId, startTimeInMillis, collectedRepairStats);
|
||||
}
|
||||
else
|
||||
{
|
||||
|
|
@ -318,8 +318,8 @@ public class AutoRepair
|
|||
while (retryCount <= config.getRepairMaxRetries(repairType))
|
||||
{
|
||||
RepairCoordinator task = repairState.getRepairRunnable(keyspaceName,
|
||||
Lists.newArrayList(curRepairAssignment.getTableNames()),
|
||||
ranges, primaryRangeOnly);
|
||||
Lists.newArrayList(curRepairAssignment.getTableNames()),
|
||||
ranges, primaryRangeOnly);
|
||||
RepairProgressListener listener = new RepairProgressListener(repairType);
|
||||
task.addProgressListener(listener);
|
||||
f = repairRunnableExecutors.get(repairType).submit(task);
|
||||
|
|
@ -449,7 +449,7 @@ public class AutoRepair
|
|||
}
|
||||
|
||||
private void cleanupAndUpdateStats(RepairTurn turn, AutoRepairConfig.RepairType repairType, AutoRepairState repairState, UUID myId,
|
||||
long startTime, CollectedRepairStats collectedRepairStats) throws InterruptedException
|
||||
long startTimeInMillis, CollectedRepairStats collectedRepairStats) throws InterruptedException
|
||||
{
|
||||
//if it was due to priority then remove it now
|
||||
if (turn == MY_TURN_DUE_TO_PRIORITY)
|
||||
|
|
@ -457,12 +457,19 @@ public class AutoRepair
|
|||
logger.info("Remove current host from priority list");
|
||||
AutoRepairUtils.removePriorityStatus(repairType, myId);
|
||||
}
|
||||
|
||||
long repairScheduleElapsedInMillis = timeFunc.get() - startTimeInMillis;
|
||||
if (repairScheduleElapsedInMillis < SLEEP_IF_REPAIR_FINISHES_QUICKLY.toMilliseconds())
|
||||
{
|
||||
//If repair finished quickly, happens for Cassndra cluster with empty (or tiny) data, in such cases,
|
||||
//wait for some duration so that the JMX metrics can detect the repairInProgress
|
||||
logger.info("Wait for {}ms for repair type {}.", SLEEP_IF_REPAIR_FINISHES_QUICKLY.toMilliseconds() - repairScheduleElapsedInMillis, repairType);
|
||||
Thread.sleep(SLEEP_IF_REPAIR_FINISHES_QUICKLY.toMilliseconds() - repairScheduleElapsedInMillis);
|
||||
}
|
||||
repairState.setFailedTokenRangesCount(collectedRepairStats.failedTokenRanges);
|
||||
repairState.setSucceededTokenRangesCount(collectedRepairStats.succeededTokenRanges);
|
||||
repairState.setSkippedTokenRangesCount(collectedRepairStats.skippedTokenRanges);
|
||||
repairState.setSkippedTablesCount(collectedRepairStats.skippedTables);
|
||||
repairState.setNodeRepairTimeInSec((int) TimeUnit.MILLISECONDS.toSeconds(timeFunc.get() - startTime));
|
||||
repairState.setNodeRepairTimeInSec((int) TimeUnit.MILLISECONDS.toSeconds(repairScheduleElapsedInMillis));
|
||||
long timeInHours = TimeUnit.SECONDS.toHours(repairState.getNodeRepairTimeInSec());
|
||||
logger.info("Local {} repair time {} hour(s), stats: repairKeyspaceCount {}, " +
|
||||
"repairTokenRangesSuccessCount {}, repairTokenRangesFailureCount {}, " +
|
||||
|
|
@ -477,13 +484,7 @@ public class AutoRepair
|
|||
TimeUnit.SECONDS.toDays(repairState.getClusterRepairTimeInSec()));
|
||||
}
|
||||
repairState.setLastRepairTime(timeFunc.get());
|
||||
if (timeInHours == 0 && SLEEP_IF_REPAIR_FINISHES_QUICKLY.toSeconds() > 0)
|
||||
{
|
||||
//If repair finished quickly, happens for an empty instance, in such case
|
||||
//wait for some duration so that the JMX metrics can detect the repairInProgress
|
||||
logger.info("Wait for {} for repair type {}.", SLEEP_IF_REPAIR_FINISHES_QUICKLY, repairType);
|
||||
Thread.sleep(SLEEP_IF_REPAIR_FINISHES_QUICKLY.toMilliseconds());
|
||||
}
|
||||
|
||||
repairState.setRepairInProgress(false);
|
||||
AutoRepairUtils.updateFinishAutoRepairHistory(repairType, myId, timeFunc.get());
|
||||
}
|
||||
|
|
|
|||
|
|
@ -36,6 +36,7 @@ import org.apache.cassandra.distributed.test.TestBaseImpl;
|
|||
import org.apache.cassandra.repair.autorepair.AutoRepair;
|
||||
import org.apache.cassandra.repair.autorepair.AutoRepairConfig;
|
||||
import org.apache.cassandra.service.AutoRepairService;
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
|
||||
import static org.hamcrest.Matchers.greaterThan;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
|
@ -114,7 +115,8 @@ public class AutoRepairSchedulerDisallowParallelReplicaRepairAcrossSchedulesTest
|
|||
cluster.forEach(i -> i.runOnInstance(() -> {
|
||||
// Expect contention on incremental repair across schedules
|
||||
AutoRepairMetrics incrementalMetrics = AutoRepairMetricsManager.getMetrics(AutoRepairConfig.RepairType.INCREMENTAL);
|
||||
Util.spinAssert("AutoRepair has not observed any replica contention in INCREMENTAL repair",
|
||||
Util.spinAssert(String.format("%s: AutoRepair has not observed any replica contention in INCREMENTAL repair",
|
||||
FBUtilities.getJustBroadcastAddress().toString()),
|
||||
greaterThan(0L),
|
||||
incrementalMetrics.repairDelayedBySchedule::getCount,
|
||||
5,
|
||||
|
|
|
|||
|
|
@ -44,6 +44,7 @@ import org.apache.cassandra.distributed.test.TestBaseImpl;
|
|||
import org.apache.cassandra.repair.autorepair.AutoRepair;
|
||||
import org.apache.cassandra.repair.autorepair.AutoRepairConfig;
|
||||
import org.apache.cassandra.service.AutoRepairService;
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
|
||||
import static org.apache.cassandra.schema.SchemaConstants.DISTRIBUTED_KEYSPACE_NAME;
|
||||
import static org.hamcrest.Matchers.greaterThan;
|
||||
|
|
@ -78,11 +79,11 @@ public class AutoRepairSchedulerTest extends TestBaseImpl
|
|||
ImmutableMap.of(
|
||||
"initial_scheduler_delay", "5s",
|
||||
"enabled", "true",
|
||||
"parallel_repair_count", "2",
|
||||
"parallel_repair_count", "3",
|
||||
// Allow parallel replica repair to allow replicas
|
||||
// to execute full repair at same time.
|
||||
"allow_parallel_replica_repair", "true",
|
||||
"min_repair_interval", "15s"),
|
||||
"min_repair_interval", "5s"),
|
||||
AutoRepairConfig.RepairType.INCREMENTAL.getConfigName(),
|
||||
ImmutableMap.of(
|
||||
"initial_scheduler_delay", "5s",
|
||||
|
|
@ -137,19 +138,21 @@ public class AutoRepairSchedulerTest extends TestBaseImpl
|
|||
|
||||
// validate that the repair ran on all nodes
|
||||
cluster.forEach(i -> i.runOnInstance(() -> {
|
||||
String broadcastAddress = FBUtilities.getJustBroadcastAddress().toString();
|
||||
|
||||
// Reduce sleeping if repair finishes quickly to speed up test but make it non-zero to provoke some
|
||||
// contention.
|
||||
AutoRepair.SLEEP_IF_REPAIR_FINISHES_QUICKLY = new DurationSpec.IntSecondsBound("1s");
|
||||
AutoRepair.SLEEP_IF_REPAIR_FINISHES_QUICKLY = new DurationSpec.IntSecondsBound("2s");
|
||||
|
||||
AutoRepairMetrics incrementalMetrics = AutoRepairMetricsManager.getMetrics(AutoRepairConfig.RepairType.INCREMENTAL);
|
||||
Util.spinAssert("AutoRepair has not yet completed one INCREMENTAL repair cycle",
|
||||
Util.spinAssert(String.format("%s: AutoRepair has not yet completed one INCREMENTAL repair cycle", broadcastAddress),
|
||||
greaterThan(0L),
|
||||
() -> incrementalMetrics.nodeRepairTimeInSec.getValue().longValue(),
|
||||
5,
|
||||
TimeUnit.MINUTES);
|
||||
|
||||
// Expect some contention on incremental repair.
|
||||
Util.spinAssert("AutoRepair has not observed any replica contention in INCREMENTAL repair",
|
||||
Util.spinAssert(String.format("%s: AutoRepair has not observed any replica contention in INCREMENTAL repair", broadcastAddress),
|
||||
greaterThan(0L),
|
||||
incrementalMetrics.repairDelayedByReplica::getCount,
|
||||
5,
|
||||
|
|
@ -159,7 +162,7 @@ public class AutoRepairSchedulerTest extends TestBaseImpl
|
|||
assertEquals(0L, incrementalMetrics.repairDelayedBySchedule.getCount());
|
||||
|
||||
AutoRepairMetrics fullMetrics = AutoRepairMetricsManager.getMetrics(AutoRepairConfig.RepairType.FULL);
|
||||
Util.spinAssert("AutoRepair has not yet completed one FULL repair cycle",
|
||||
Util.spinAssert(String.format("%s: AutoRepair has not yet completed one FULL repair cycle", broadcastAddress),
|
||||
greaterThan(0L),
|
||||
() -> fullMetrics.nodeRepairTimeInSec.getValue().longValue(),
|
||||
5,
|
||||
|
|
|
|||
Loading…
Reference in New Issue