natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Backport CASSANDRA-20469 (Paxos repair interrupts running transactions) to 4.1 and 5.0 

patch by Ariel Weisberg; reviewed by Benedict Elliott Smith for CASSANDRA-20585
This commit is contained in:
Ariel Weisberg 2025-05-29 16:56:22 -04:00
parent eb4e79500e
commit 88ca4f8d9a
5 changed files with 61 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/java/org/apache/cassandra/config/Config.java
View File

@ -1264,4 +1264,6 @@ public class Config
// 3.0 Cassandra Driver has its "read" timeout set to 12 seconds. Our recommendation is match this.
public DurationSpec.LongMillisecondsBound native_transport_timeout = new DurationSpec.LongMillisecondsBound("12000ms");
public boolean enforce_native_deadline_for_hints = false;
public boolean paxos_repair_race_wait = true;
}

11
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -4638,4 +4638,15 @@ public class DatabaseDescriptor
{
conf.reject_out_of_token_range_requests = enabled;
}
public static boolean getPaxosRepairRaceWait()
{
return conf.paxos_repair_race_wait;
}
@VisibleForTesting
public static void setPaxosRepairRaceWait(boolean paxosRepairRaceWait)
{
conf.paxos_repair_race_wait = paxosRepairRaceWait;
}
}

12
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -7153,4 +7153,16 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
DatabaseDescriptor.setEnforceNativeDeadlineForHints(value);
}
@Override
public void setPaxosRepairRaceWait(boolean paxosRepairRaceWait)
{
DatabaseDescriptor.setPaxosRepairRaceWait(paxosRepairRaceWait);
}
@Override
public boolean getPaxosRepairRaceWait()
{
return DatabaseDescriptor.getPaxosRepairRaceWait();
}
}

4
src/java/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -1103,4 +1103,8 @@ public interface StorageServiceMBean extends NotificationEmitter
* e.g. keyspace_name -> [reads, writes, paxos].
*/
Map<String, long[]> getOutOfRangeOperationCounts();
void setPaxosRepairRaceWait(boolean paxosRepairCoordinatorWait);
boolean getPaxosRepairRaceWait();
}

34
src/java/org/apache/cassandra/service/paxos/cleanup/PaxosCleanupLocalCoordinator.java
View File

@ -24,6 +24,7 @@ import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.Uninterruptibles;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -39,9 +40,15 @@ import org.apache.cassandra.service.paxos.AbstractPaxosRepair;
import org.apache.cassandra.service.paxos.PaxosRepair;
import org.apache.cassandra.service.paxos.PaxosState;
import org.apache.cassandra.service.paxos.uncommitted.UncommittedPaxosKey;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.CloseableIterator;
import org.apache.cassandra.utils.concurrent.AsyncFuture;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.cassandra.config.DatabaseDescriptor.getCasContentionTimeout;
import static org.apache.cassandra.config.DatabaseDescriptor.getWriteRpcTimeout;
import static org.apache.cassandra.service.paxos.cleanup.PaxosCleanupSession.TIMEOUT_NANOS;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
@ -126,8 +133,10 @@ public class PaxosCleanupLocalCoordinator extends AsyncFuture<PaxosCleanupRespon
return;
}
long txnTimeoutMicros = Math.max(getCasContentionTimeout(MICROSECONDS), getWriteRpcTimeout(MICROSECONDS));
boolean waitForCoordinator = DatabaseDescriptor.getPaxosRepairRaceWait();
while (inflight.size() < parallelism && uncommittedIter.hasNext())
repairKey(uncommittedIter.next());
repairKey(uncommittedIter.next(), txnTimeoutMicros, waitForCoordinator);
}
@ -135,7 +144,7 @@ public class PaxosCleanupLocalCoordinator extends AsyncFuture<PaxosCleanupRespon
finish();
}
private boolean repairKey(UncommittedPaxosKey uncommitted)
private boolean repairKey(UncommittedPaxosKey uncommitted, long txnTimeoutMicros, boolean waitForCoordinator)
{
logger.trace("repairing {}", uncommitted);
Preconditions.checkState(!inflight.containsKey(uncommitted.getKey()));
@ -146,6 +155,9 @@ public class PaxosCleanupLocalCoordinator extends AsyncFuture<PaxosCleanupRespon
if (consistency == null)
return false;
if (waitForCoordinator)
maybeWaitForOriginalCoordinator(uncommitted, txnTimeoutMicros);
inflight.put(uncommitted.getKey(), tableRepairs.startOrGetOrQueue(uncommitted.getKey(), uncommitted.ballot(), uncommitted.getConsistencyLevel(), table, result -> {
if (result.wasSuccessful())
onKeyFinish(uncommitted.getKey());
@ -155,6 +167,24 @@ public class PaxosCleanupLocalCoordinator extends AsyncFuture<PaxosCleanupRespon
return true;
}
/**
* Wait to repair things that are still potentially executing at the original coordinator to avoid
* causing timeouts. This should only have to happen at most a few times when the repair starts
*/
private static void maybeWaitForOriginalCoordinator(UncommittedPaxosKey uncommitted, long txnTimeoutMicros)
{
long nowMicros = MILLISECONDS.toMicros(Clock.Global.currentTimeMillis());
long ballotElapsedMicros = nowMicros - uncommitted.ballot().unixMicros();
if (ballotElapsedMicros < 0 && Math.abs(ballotElapsedMicros) > SECONDS.toMicros(1))
logger.warn("Encountered ballot that is more than 1 second in the future, is there a clock sync issue? {}", uncommitted.ballot());
if (ballotElapsedMicros < txnTimeoutMicros)
{
long sleepMicros = txnTimeoutMicros - ballotElapsedMicros;
logger.info("Paxos auto repair encountered a potentially in progress ballot, sleeping {}us to allow the in flight operation to finish", sleepMicros);
Uninterruptibles.sleepUninterruptibly(sleepMicros, MICROSECONDS);
}
}
private synchronized void onKeyFinish(DecoratedKey key)
{
if (!inflight.containsKey(key))