From cf60eb8672bf162a47b55ed31ff785440fa53840 Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Thu, 3 Apr 2025 11:31:28 -0700 Subject: [PATCH] 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 --- CHANGES.txt | 1 + .../cassandra/service/StorageProxy.java | 2 +- .../service/paxos/v1/PrepareCallback.java | 29 +++++- .../upgrade/MixedModePaxosTTLTest.java | 94 +++++++++++++++++++ .../upgrade/MixedModePaxosTestBase.java | 55 +++++++++-- 5 files changed, 170 insertions(+), 11 deletions(-) create mode 100644 test/distributed/org/apache/cassandra/distributed/upgrade/MixedModePaxosTTLTest.java diff --git a/CHANGES.txt b/CHANGES.txt index 0dfb6d1c0b..d1aa662fe9 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index d4a61b996d..41703c439c 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -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 missingMRC = summary.replicasMissingMostRecentCommit(); + Iterable missingMRC = summary.replicasMissingMostRecentCommit(metadata); 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/v1/PrepareCallback.java b/src/java/org/apache/cassandra/service/paxos/v1/PrepareCallback.java index 4aedb6d63d..315c7abb9f 100644 --- a/src/java/org/apache/cassandra/service/paxos/v1/PrepareCallback.java +++ b/src/java/org/apache/cassandra/service/paxos/v1/PrepareCallback.java @@ -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 { @@ -79,8 +86,28 @@ public class PrepareCallback extends AbstractPaxosCallback latch.decrement(); } - public Iterable replicasMissingMostRecentCommit() + public Iterable 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))); } } diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModePaxosTTLTest.java b/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModePaxosTTLTest.java new file mode 100644 index 0000000000..f7e2d9272c --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModePaxosTTLTest.java @@ -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(); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModePaxosTestBase.java b/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModePaxosTestBase.java index 12e7b96561..e9e16f738a 100644 --- a/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModePaxosTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModePaxosTestBase.java @@ -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); } } }