mirror of https://github.com/apache/cassandra
Fix mixed mode paxos hang
- reinstates the mrc check from CASSANDRA-12043 for legacy paxos purging Patch by Blake Eggleston; Reviewed by Ariel Weisberg for CASSANDRA-20514
This commit is contained in:
Blake Eggleston
parent
147a224a88
commit
cf60eb8672
|
|
@ -1,4 +1,5 @@
|
|||
4.1.9
|
||||
* Fix mixed mode paxos ttl commit hang (CASSANDRA-20514)
|
||||
* Fix paxos mixed mode infinite loop (CASSANDRA-20493)
|
||||
* Optionally skip exception logging on invalid legacy protocol magic exception (CASSANDRA-19483)
|
||||
* Fix SimpleClient ability to release acquired capacity (CASSANDRA-20202)
|
||||
|
|
|
|||
|
|
@ -652,7 +652,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<InetAddressAndPort> missingMRC = summary.replicasMissingMostRecentCommit();
|
||||
Iterable<InetAddressAndPort> missingMRC = summary.replicasMissingMostRecentCommit(metadata);
|
||||
if (Iterables.size(missingMRC) > 0)
|
||||
{
|
||||
Tracing.trace("Repairing replicas that missed the most recent commit");
|
||||
|
|
|
|||
|
|
@ -19,11 +19,16 @@
|
|||
package org.apache.cassandra.service.paxos.v1;
|
||||
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import com.google.common.collect.Iterables;
|
||||
|
||||
import org.apache.cassandra.config.Config;
|
||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.db.SystemKeyspace;
|
||||
import org.apache.cassandra.locator.InetAddressAndPort;
|
||||
import org.apache.cassandra.schema.TableMetadata;
|
||||
import org.apache.cassandra.db.ConsistencyLevel;
|
||||
|
|
@ -35,6 +40,8 @@ import org.apache.cassandra.net.Message;
|
|||
import org.apache.cassandra.service.paxos.Commit;
|
||||
import org.apache.cassandra.service.paxos.PrepareResponse;
|
||||
import org.apache.cassandra.transport.Dispatcher;
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
import org.apache.cassandra.utils.UUIDGen;
|
||||
|
||||
public class PrepareCallback extends AbstractPaxosCallback<PrepareResponse>
|
||||
{
|
||||
|
|
@ -79,8 +86,28 @@ public class PrepareCallback extends AbstractPaxosCallback<PrepareResponse>
|
|||
latch.decrement();
|
||||
}
|
||||
|
||||
public Iterable<InetAddressAndPort> replicasMissingMostRecentCommit()
|
||||
public Iterable<InetAddressAndPort> replicasMissingMostRecentCommit(TableMetadata metadata)
|
||||
{
|
||||
/**
|
||||
* this check is only needed for mixed mode operation with 4.0 and can be removed once upgrade support dropped
|
||||
* see the comment in {@link org.apache.cassandra.distributed.upgrade.MixedModePaxosTTLTest} for a full explanation.
|
||||
*/
|
||||
if (DatabaseDescriptor.paxosStatePurging() == Config.PaxosStatePurging.legacy)
|
||||
{
|
||||
// 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 an MRC that is too old, i.e. older than the TTL we set
|
||||
// on paxos tables. For such an old commit, we rely on hints and repair to ensure the commit has indeed been
|
||||
// propagated to all nodes.
|
||||
long paxosTtlSec = SystemKeyspace.legacyPaxosTtlSec(metadata);
|
||||
if (TimeUnit.MICROSECONDS.toSeconds(mostRecentCommit.ballot.unixMicros()) + paxosTtlSec < FBUtilities.nowInSeconds())
|
||||
return Collections.emptySet();
|
||||
}
|
||||
|
||||
return Iterables.filter(commitsByReplica.keySet(), inetAddress -> (!commitsByReplica.get(inetAddress).ballot.equals(mostRecentCommit.ballot)));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,94 @@
|
|||
/*
|
||||
* 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.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.apache.cassandra.distributed.api.ConsistencyLevel;
|
||||
import org.apache.cassandra.distributed.api.Feature;
|
||||
import org.apache.cassandra.distributed.upgrade.MixedModePaxosTestBase.FakePaxosHelper;
|
||||
|
||||
import static java.lang.String.format;
|
||||
|
||||
public class MixedModePaxosTTLTest extends UpgradeTestBase
|
||||
{
|
||||
/**
|
||||
* Tests the mixed mode paxos loop bug in CASSANDRA-20514
|
||||
*
|
||||
* CEP-14 changed the ttl behavior of legacy paxos state to expire based off the ballot time of the operation being
|
||||
* persisted, not the time a commit is persisted. This eliminated the race addressed by CASSANDRA-12043, and so the
|
||||
* check it added to the most recent commit prepare logic was removed.
|
||||
*
|
||||
* When operating in mixed mode though, this can still be a problem. If a 4.1 or higher node is coordinating a paxos
|
||||
* operation with 2 or more replicas on 4.0 or lower, this race becomes a problem again. You need 3 things to make
|
||||
* this an infinite loop
|
||||
* 1. a 4.1 node coordinating a paxos operation with 2x 4.0 replicas
|
||||
* 2. replica A) a 4.0 node returns a most recent commit for a ballot that's could have been ttld
|
||||
* 3. replica B) a 4.0 node has ttl'd that mrc AND converted the ttld cells into tombstones
|
||||
*
|
||||
* The 4.1 coordinator receives the mrc from replica A, but since it no longer disregards missing most recent commits
|
||||
* past the ttl window, it sends the "missing" commit to replica B. Since replica B now has a tombstone for that mrc,
|
||||
* and tombstones win when reconciled with live cells, even ones with ttls, the commit is a noop and it continues
|
||||
* to report nothing for its mrc value when the coordinator restarts the prepare phase. This loops until the query
|
||||
* times out.
|
||||
*/
|
||||
@Test
|
||||
public void legacyExpiredStateTest() throws Throwable
|
||||
{
|
||||
String keyspace = "ks";
|
||||
String table = "tbl";
|
||||
int gcGrace = 60*60*24; // 1 day
|
||||
int key = 100; // hashes to nodes 2 & 3 w/ murmur @ RF=2
|
||||
new TestCase()
|
||||
.withConfig(c -> c.with(Feature.GOSSIP, Feature.NETWORK).set("cas_contention_timeout", "500ms"))
|
||||
.nodes(3)
|
||||
.nodesToUpgrade(1)
|
||||
.singleUpgrade(v40)
|
||||
.setup(cluster -> {
|
||||
cluster.schemaChange(format("CREATE KEYSPACE %s WITH REPLICATION={'class': 'SimpleStrategy', 'replication_factor': '2'}", keyspace));
|
||||
cluster.schemaChange(format("CREATE TABLE %s.%s (k int primary key, v int) " +
|
||||
"WITH gc_grace_seconds=%s", keyspace, table, gcGrace));
|
||||
})
|
||||
.runAfterClusterUpgrade(cluster -> {
|
||||
// disable compaction to prevent paxos state from being purged
|
||||
cluster.forEach(instance -> instance.nodetool("disableautocompaction"));
|
||||
|
||||
long ballotMicros = TimeUnit.MILLISECONDS.toMicros(System.currentTimeMillis());
|
||||
ballotMicros -= TimeUnit.SECONDS.toMicros(gcGrace + 10);
|
||||
FakePaxosHelper helper = FakePaxosHelper.create(cluster.coordinator(1), keyspace, table, key, gcGrace, ballotMicros);
|
||||
|
||||
// confirm none of the nodes have paxos state
|
||||
for (int i = 1; i <= cluster.size(); i++)
|
||||
helper.assertNoPaxosData(cluster.coordinator(i));
|
||||
|
||||
// save a tombstoned commit to one node to simulate expired cells being converted to tombstones
|
||||
helper.tombstoneCommit(cluster.coordinator(2));
|
||||
|
||||
// insert paxos state and confirm it hasn't ttl'd yet
|
||||
helper.saveCommit(cluster.coordinator(3));
|
||||
helper.assertPaxosData(cluster.coordinator(3));
|
||||
|
||||
// paxos operation should not timeout
|
||||
cluster.coordinator(1).execute(format("SELECT * FROM %s.%s WHERE k=%s", keyspace, table, key), ConsistencyLevel.SERIAL);
|
||||
})
|
||||
.run();
|
||||
}
|
||||
}
|
||||
|
|
@ -76,6 +76,7 @@ public abstract class MixedModePaxosTestBase extends UpgradeTestBase
|
|||
String keyspace = KEYSPACE;
|
||||
String table = "tbl";
|
||||
int gcGrace = 10;
|
||||
int key = 1;
|
||||
new TestCase()
|
||||
.withConfig(c -> c.with(Feature.GOSSIP, Feature.NETWORK))
|
||||
.nodes(2)
|
||||
|
|
@ -91,11 +92,11 @@ public abstract class MixedModePaxosTestBase extends UpgradeTestBase
|
|||
|
||||
// insert a ttl'd committed paxos state
|
||||
long ballotMicros = TimeUnit.NANOSECONDS.toMicros(System.currentTimeMillis());
|
||||
FakePaxosHelper helper = FakePaxosHelper.create(cluster.coordinator(1), keyspace, table, gcGrace, ballotMicros);
|
||||
FakePaxosHelper helper = FakePaxosHelper.create(cluster.coordinator(1), keyspace, table, key, gcGrace, ballotMicros);
|
||||
|
||||
// confirm none of the nodes have paxos state
|
||||
for (int i = 1; i <= cluster.size(); i++)
|
||||
Assert.assertEquals(0, cluster.coordinator(i).execute("SELECT * FROM system.paxos", ConsistencyLevel.ONE).length);
|
||||
helper.assertNoPaxosData(cluster.coordinator(i));
|
||||
|
||||
|
||||
// save commit to both nodes
|
||||
|
|
@ -109,11 +110,11 @@ public abstract class MixedModePaxosTestBase extends UpgradeTestBase
|
|||
Thread.sleep(TimeUnit.SECONDS.toMillis(gcGrace * 2));
|
||||
|
||||
// confirm paxos state has ttld
|
||||
Assert.assertEquals(0, cluster.coordinator(1).execute("SELECT * FROM system.paxos", ConsistencyLevel.ONE).length);
|
||||
Assert.assertEquals(0, cluster.coordinator(2).execute("SELECT * FROM system.paxos", ConsistencyLevel.ONE).length);
|
||||
helper.assertNoPaxosData(cluster.coordinator(1));
|
||||
helper.assertNoPaxosData(cluster.coordinator(2));
|
||||
|
||||
// paxos operation should not timeout
|
||||
cluster.coordinator(upgradedCoordinator() ? 1 : 2).execute(format("SELECT * FROM %s.%s WHERE k=1", keyspace, table), ConsistencyLevel.SERIAL);
|
||||
cluster.coordinator(upgradedCoordinator() ? 1 : 2).execute(format("SELECT * FROM %s.%s WHERE k=%s", keyspace, table, key), ConsistencyLevel.SERIAL);
|
||||
})
|
||||
.run();
|
||||
}
|
||||
|
|
@ -133,14 +134,14 @@ public abstract class MixedModePaxosTestBase extends UpgradeTestBase
|
|||
@Test
|
||||
public void bothAwareTTldPaxosStateTest() throws Throwable
|
||||
{
|
||||
ttldPaxosStateTest(true, false);
|
||||
ttldPaxosStateTest(true, true);
|
||||
}
|
||||
|
||||
/**
|
||||
* This is an upgrade test, and paxos internally limits ttls to 3 hours, so we have to manually save commits in
|
||||
* the paxos table to get entries ttl'd in a reasonable amount of time
|
||||
*/
|
||||
private static class FakePaxosHelper
|
||||
static class FakePaxosHelper
|
||||
{
|
||||
static final int current_version = MessagingService.current_version;
|
||||
static final int version_40a = MessagingService.VERSION_40;
|
||||
|
|
@ -181,6 +182,21 @@ public abstract class MixedModePaxosTestBase extends UpgradeTestBase
|
|||
return PartitionUpdate.toBytes(update, version);
|
||||
}
|
||||
|
||||
private Object[][] paxosData(ICoordinator coordinator)
|
||||
{
|
||||
return coordinator.execute("SELECT * FROM system.paxos WHERE row_key = ? AND cf_id = ?", ConsistencyLevel.ONE, key, cfId);
|
||||
}
|
||||
|
||||
void assertNoPaxosData(ICoordinator coordinator)
|
||||
{
|
||||
Assert.assertEquals(0, paxosData(coordinator).length);
|
||||
}
|
||||
|
||||
void assertPaxosData(ICoordinator coordinator)
|
||||
{
|
||||
Assert.assertEquals(1, paxosData(coordinator).length);
|
||||
}
|
||||
|
||||
void saveCommit(ICoordinator coordinator)
|
||||
{
|
||||
String cql = "UPDATE system.paxos USING TIMESTAMP ? AND TTL ? SET proposal_ballot = null, proposal = null, most_recent_commit_at = ?, most_recent_commit = ?, most_recent_commit_version = ? WHERE row_key = ? AND cf_id = ?";
|
||||
|
|
@ -194,10 +210,31 @@ public abstract class MixedModePaxosTestBase extends UpgradeTestBase
|
|||
cfId);
|
||||
}
|
||||
|
||||
public static FakePaxosHelper create(ICoordinator coordinator, String keyspace, String table, int ttl, long ballotMicros)
|
||||
void tombstoneCommit(ICoordinator coordinator)
|
||||
{
|
||||
String cql = "DELETE proposal_ballot, proposal, most_recent_commit_at, most_recent_commit, most_recent_commit_version FROM system.paxos USING TIMESTAMP ? WHERE row_key = ? AND cf_id = ?";
|
||||
coordinator.execute(cql, ConsistencyLevel.ONE,
|
||||
ballotMicros,
|
||||
key,
|
||||
cfId);
|
||||
}
|
||||
|
||||
void saveCommitNoTTL(ICoordinator coordinator)
|
||||
{
|
||||
String cql = "UPDATE system.paxos USING TIMESTAMP ? SET proposal_ballot = null, proposal = null, most_recent_commit_at = ?, most_recent_commit = ?, most_recent_commit_version = ? WHERE row_key = ? AND cf_id = ?";
|
||||
coordinator.execute(cql, ConsistencyLevel.ONE,
|
||||
ballotMicros,
|
||||
ballot,
|
||||
updateBytes(version_40a),
|
||||
version_40a,
|
||||
key,
|
||||
cfId);
|
||||
}
|
||||
|
||||
public static FakePaxosHelper create(ICoordinator coordinator, String keyspace, String table, int key, int ttl, long ballotMicros)
|
||||
{
|
||||
UUID cfId = (UUID) coordinator.execute("SELECT id FROM system_schema.tables WHERE keyspace_name=? AND table_name=?", ConsistencyLevel.ONE, keyspace, table)[0][0];
|
||||
return new FakePaxosHelper(keyspace, table, cfId, 1, ttl, ballotMicros);
|
||||
return new FakePaxosHelper(keyspace, table, cfId, key, ttl, ballotMicros);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue