diff --git a/CHANGES.txt b/CHANGES.txt index 8d2062d108..9f42d98f87 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -30,6 +30,7 @@ * Always close cluster with connection in CqlRecordWriter (CASSANDRA-11553) * Fix slice queries on ordered COMPACT tables (CASSANDRA-10988) Merged from 2.1: + * Avoid stalling paxos when the paxos state expires (CASSANDRA-12043) * Remove finished incoming streaming connections from MessagingService (CASSANDRA-11854) * Don't try to get sstables for non-repairing column families (CASSANDRA-12077) * Prevent select statements with clustering key > 64k (CASSANDRA-11882) diff --git a/src/java/org/apache/cassandra/cql3/QueryProcessor.java b/src/java/org/apache/cassandra/cql3/QueryProcessor.java index fa82fa7c3d..c70267963c 100644 --- a/src/java/org/apache/cassandra/cql3/QueryProcessor.java +++ b/src/java/org/apache/cassandra/cql3/QueryProcessor.java @@ -291,7 +291,7 @@ public class QueryProcessor implements QueryHandler return QueryOptions.forInternalCalls(boundValues); } - private static ParsedStatement.Prepared prepareInternal(String query) throws RequestValidationException + public static ParsedStatement.Prepared prepareInternal(String query) throws RequestValidationException { ParsedStatement.Prepared prepared = internalStatements.get(query); if (prepared != null) @@ -340,6 +340,32 @@ public class QueryProcessor implements QueryHandler return null; } + /** + * A special version of executeInternal that takes the time used as "now" for the query in argument. + * Note that this only make sense for Selects so this only accept SELECT statements and is only useful in rare + * cases. + */ + public static UntypedResultSet executeInternalWithNow(long now, String query, Object... values) + { + try + { + ParsedStatement.Prepared prepared = prepareInternal(query); + assert prepared.statement instanceof SelectStatement; + SelectStatement select = (SelectStatement)prepared.statement; + ResultMessage result = select.executeInternal(internalQueryState(), makeInternalOptions(prepared, values), now); + assert result instanceof ResultMessage.Rows; + return UntypedResultSet.create(((ResultMessage.Rows)result).result); + } + catch (RequestExecutionException e) + { + throw new RuntimeException(e); + } + catch (RequestValidationException e) + { + throw new RuntimeException("Error validating query " + query, e); + } + } + public static UntypedResultSet resultify(String query, Row row) { return resultify(query, Collections.singletonList(row)); diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java index 20fe98240f..8820ff7823 100644 --- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java @@ -290,9 +290,13 @@ public class SelectStatement implements CQLStatement } public ResultMessage.Rows executeInternal(QueryState state, QueryOptions options) throws RequestExecutionException, RequestValidationException + { + return executeInternal(state, options, System.currentTimeMillis()); + } + + public ResultMessage.Rows executeInternal(QueryState state, QueryOptions options, long now) throws RequestExecutionException, RequestValidationException { int limit = getLimit(options); - long now = System.currentTimeMillis(); Pageable command = getPageableCommand(options, limit, now); int pageSize = getPageSize(options); diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index 74a3c7b0ac..e0d5f661b1 100644 --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java @@ -894,10 +894,10 @@ public final class SystemKeyspace return null; } - public static PaxosState loadPaxosState(ByteBuffer key, CFMetaData metadata) + public static PaxosState loadPaxosState(ByteBuffer key, CFMetaData metadata, long now) { String req = "SELECT * FROM system.%s WHERE row_key = ? AND cf_id = ?"; - UntypedResultSet results = executeInternal(String.format(req, PAXOS), key, metadata.cfId); + UntypedResultSet results = QueryProcessor.executeInternalWithNow(now, String.format(req, PAXOS), key, metadata.cfId); if (results.isEmpty()) return new PaxosState(key, metadata); UntypedResultSet.Row row = results.one(); @@ -937,7 +937,7 @@ public final class SystemKeyspace proposal.update.id()); } - private static int paxosTtl(CFMetaData metadata) + public static int paxosTtl(CFMetaData metadata) { // keep paxos state around for at least 3h return Math.max(3 * 3600, metadata.getGcGraceSeconds()); diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 5e25211a75..03dd209aa9 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -425,7 +425,7 @@ public class StorageProxy implements StorageProxyMBean // https://issues.apache.org/jira/browse/CASSANDRA-5062?focusedCommentId=13619810&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-13619810) // Since we waited for quorum nodes, if some of them haven't seen the last commit (which may just be a timing issue, but may also // mean we lost messages), we pro-actively "repair" those nodes, and retry. - Iterable missingMRC = summary.replicasMissingMostRecentCommit(); + Iterable missingMRC = summary.replicasMissingMostRecentCommit(metadata, ballotMicros); if (Iterables.size(missingMRC) > 0) { Tracing.trace("Repairing replicas that missed the most recent commit"); diff --git a/src/java/org/apache/cassandra/service/paxos/PaxosState.java b/src/java/org/apache/cassandra/service/paxos/PaxosState.java index 01e03f47c6..fde881bfff 100644 --- a/src/java/org/apache/cassandra/service/paxos/PaxosState.java +++ b/src/java/org/apache/cassandra/service/paxos/PaxosState.java @@ -63,7 +63,13 @@ public class PaxosState lock.lock(); try { - PaxosState state = SystemKeyspace.loadPaxosState(toPrepare.key, toPrepare.update.metadata()); + // When preparing, we need to use the same time as "now" (that's the time we use to decide if something + // is expired or not) accross nodes otherwise we may have a window where a Most Recent Commit shows up + // on some replica and not others during a new proposal (in StorageProxy.beginAndRepairPaxos()), and no + // amount of re-submit will fix this (because the node on which the commit has expired will have a + // tombstone that hides any re-submit). See CASSANDRA-12043 for details. + long now = UUIDGen.unixTimestamp(toPrepare.ballot); + PaxosState state = SystemKeyspace.loadPaxosState(toPrepare.key, toPrepare.update.metadata(), now); if (toPrepare.isAfter(state.promised)) { Tracing.trace("Promising ballot {}", toPrepare.ballot); @@ -98,7 +104,8 @@ public class PaxosState lock.lock(); try { - PaxosState state = SystemKeyspace.loadPaxosState(proposal.key, proposal.update.metadata()); + long now = UUIDGen.unixTimestamp(proposal.ballot); + PaxosState state = SystemKeyspace.loadPaxosState(proposal.key, proposal.update.metadata(), now); if (proposal.hasBallot(state.promised.ballot) || proposal.isAfter(state.promised)) { Tracing.trace("Accepting proposal {}", proposal); diff --git a/src/java/org/apache/cassandra/service/paxos/PrepareCallback.java b/src/java/org/apache/cassandra/service/paxos/PrepareCallback.java index ad055d0854..081f457983 100644 --- a/src/java/org/apache/cassandra/service/paxos/PrepareCallback.java +++ b/src/java/org/apache/cassandra/service/paxos/PrepareCallback.java @@ -23,6 +23,7 @@ package org.apache.cassandra.service.paxos; import java.net.InetAddress; import java.nio.ByteBuffer; +import java.util.Collections; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -33,7 +34,9 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.config.CFMetaData; +import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.net.MessageIn; +import org.apache.cassandra.utils.UUIDGen; public class PrepareCallback extends AbstractPaxosCallback { @@ -86,8 +89,21 @@ public class PrepareCallback extends AbstractPaxosCallback latch.countDown(); } - public Iterable replicasMissingMostRecentCommit() + public Iterable replicasMissingMostRecentCommit(CFMetaData metadata, long now) { + // In general, we need every replicas that have answered to the prepare (a quorum) to agree on the MRC (see + // coment in StorageProxy.beginAndRepairPaxos(), but basically we need to make sure at least a quorum of nodes + // have learn a commit before commit a new one otherwise that previous commit is not guaranteed to have reach a + // quorum and further commit may proceed on incomplete information). + // However, if that commit is too hold, it may have been expired from some of the replicas paxos table (we don't + // keep the paxos state forever or that could grow unchecked), and we could end up in some infinite loop as + // explained on CASSANDRA-12043. To avoid that, we ignore a MRC that is too old, i.e. older than the TTL we set + // on paxos tables. For such old commit, we rely on hints and repair to ensure the commit has indeed be + // propagated to all nodes. + long paxosTtlMicros = SystemKeyspace.paxosTtl(metadata) * 1000 * 1000; + if (UUIDGen.microsTimestamp(mostRecentCommit.ballot) + paxosTtlMicros < now) + return Collections.emptySet(); + return Iterables.filter(commitsByReplica.keySet(), new Predicate() { public boolean apply(InetAddress inetAddress)