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

Merge branch 'cassandra-4.1' into trunk

This commit is contained in:
Jon Meredith 2023-05-03 14:42:57 -06:00
parent 73da05f83b 602ffcbf3e
commit ccf3789219
1 changed files with 123 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

248
test/distributed/org/apache/cassandra/distributed/test/PaxosRepair2Test.java
View File

@ -18,19 +18,16 @@
package org.apache.cassandra.distributed.test;
import java.net.InetAddress;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterators;
import com.google.common.collect.Sets;
import org.awaitility.Awaitility;
import org.junit.Assert;
import org.junit.Test;
import org.slf4j.Logger;
@ -69,7 +66,6 @@ import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.exceptions.CasWriteTimeoutException;
import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.repair.RepairParallelism;
@ -107,6 +103,7 @@ public class PaxosRepair2Test extends TestBaseImpl
{
private static final Logger logger = LoggerFactory.getLogger(PaxosRepair2Test.class);
private static final String TABLE = "tbl";
public static final String OFFSETTABLE_CLOCK_NAME = OffsettableClock.class.getName();
static
{
@ -126,19 +123,6 @@ public class PaxosRepair2Test extends TestBaseImpl
return uncommitted;
}
private static void assertAllAlive(Cluster cluster)
{
Set<InetAddress> allEndpoints = cluster.stream().map(i -> i.broadcastAddress().getAddress()).collect(Collectors.toSet());
cluster.stream().forEach(instance -> {
instance.runOnInstance(() -> {
ImmutableSet<InetAddressAndPort> endpoints = Gossiper.instance.getEndpoints();
Assert.assertEquals(allEndpoints, endpoints);
for (InetAddressAndPort endpoint : endpoints)
Assert.assertTrue(FailureDetector.instance.isAlive(endpoint));
});
});
}
private static void assertUncommitted(IInvokableInstance instance, String ks, String table, int expected)
{
Assert.assertEquals(expected, getUncommitted(instance, ks, table));
@ -259,16 +243,15 @@ public class PaxosRepair2Test extends TestBaseImpl
ClusterUtils.stopUnchecked(cluster.get(3));
InetAddressAndPort node3 = InetAddressAndPort.getByAddress(cluster.get(3).broadcastAddress());
for (int i = 0; i < 10; i++)
{
if (!cluster.get(1).callOnInstance(() -> FailureDetector.instance.isAlive(node3)))
break;
}
// make sure node1 knows node3 is down
Awaitility.waitAtMost(1,TimeUnit.MINUTES).until(
() -> !cluster.get(1).callOnInstance(() -> FailureDetector.instance.isAlive(node3)));
repair(cluster, KEYSPACE, TABLE, true);
for (int i = 0; i < cluster.size() - 1; i++)
{
cluster.get(i + 1).runOnInstance(() -> {
Assert.assertFalse(CassandraRelevantProperties.CLOCK_GLOBAL.isPresent());
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(TABLE);
DecoratedKey key = cfs.decorateKey(ByteBufferUtil.bytes(1));
Assert.assertTrue(FBUtilities.getBroadcastAddressAndPort().toString(), Commit.isAfter(staleBallot, cfs.getPaxosRepairLowBound(key)));
@ -356,6 +339,11 @@ public class PaxosRepair2Test extends TestBaseImpl
.set("truncate_request_timeout_in_ms", 1000L)))
)
{
cluster.forEach(i -> {
Assert.assertFalse(CassandraRelevantProperties.CLOCK_GLOBAL.isPresent());
Assert.assertEquals("1", System.getProperty("cassandra.auto_repair_frequency_seconds"));
Assert.assertEquals("true", System.getProperty("cassandra.disable_paxos_auto_repairs"));
});
cluster.schemaChange("CREATE TABLE " + KEYSPACE + '.' + TABLE + " (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
ClusterUtils.stopUnchecked(cluster.get(3));
cluster.verbs(Verb.PAXOS_COMMIT_REQ).drop();
@ -413,6 +401,7 @@ public class PaxosRepair2Test extends TestBaseImpl
repair(cluster, KEYSPACE, TABLE);
cluster.forEach(i -> i.runOnInstance(() -> {
Assert.assertFalse(CassandraRelevantProperties.CLOCK_GLOBAL.isPresent());
compactPaxos();
Map<Integer, PaxosRow> rows = getPaxosRows();
Assert.assertEquals(Sets.newHashSet(1), rows.keySet());
@ -471,116 +460,125 @@ public class PaxosRepair2Test extends TestBaseImpl
@Test
public void legacyPurgeRepairLoop() throws Exception
{
CassandraRelevantProperties.CLOCK_GLOBAL.setString(OffsettableClock.class.getName());
try (Cluster cluster = init(Cluster.create(3, cfg -> cfg
.set("paxos_variant", "v2")
.set("paxos_state_purging", "legacy")
.set("paxos_purge_grace_period", "0s")
.set("truncate_request_timeout_in_ms", 1000L)))
)
try
{
int ttl = 3*3600;
cluster.schemaChange("CREATE TABLE " + KEYSPACE + '.' + TABLE + " (pk int, ck int, v int, PRIMARY KEY (pk, ck)) WITH gc_grace_seconds=" + ttl);
CassandraRelevantProperties.CLOCK_GLOBAL.setString(OFFSETTABLE_CLOCK_NAME);
try (Cluster cluster = init(Cluster.create(3, cfg -> cfg
.set("paxos_variant", "v2")
.set("paxos_state_purging", "legacy")
.set("paxos_purge_grace_period", "0s")
.set("truncate_request_timeout_in_ms", 1000L)))
)
{
cluster.forEach(i -> Assert.assertEquals(OFFSETTABLE_CLOCK_NAME, CassandraRelevantProperties.CLOCK_GLOBAL.getString()));
int ttl = 3 * 3600;
cluster.schemaChange("CREATE TABLE " + KEYSPACE + '.' + TABLE + " (pk int, ck int, v int, PRIMARY KEY (pk, ck)) WITH gc_grace_seconds=" + ttl);
// prepare an operation ttl + 1 hour into the past on a single node
cluster.forEach(instance -> {
instance.runOnInstance(() -> {
backdateTimestamps(ttl + 3600);
});
});
cluster.filters().inbound().to(1, 2).drop();
assertTimeout(() -> cluster.coordinator(3).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, ck, v) VALUES (400, 2, 2) IF NOT EXISTS", ConsistencyLevel.QUORUM));
Ballot oldBallot = Ballot.fromUuid(cluster.get(3).callOnInstance(() -> {
TableMetadata cfm = Schema.instance.getTableMetadata(KEYSPACE, TABLE);
DecoratedKey dk = cfm.partitioner.decorateKey(ByteBufferUtil.bytes(400));
try (PaxosState state = PaxosState.get(dk, cfm))
{
return state.currentSnapshot().promised.asUUID();
}
}));
assertUncommitted(cluster.get(1), KEYSPACE, TABLE, 0);
assertUncommitted(cluster.get(2), KEYSPACE, TABLE, 0);
assertUncommitted(cluster.get(3), KEYSPACE, TABLE, 1);
// commit an operation just over ttl in the past on the other nodes
cluster.filters().reset();
cluster.filters().inbound().to(2).drop();
cluster.forEach(instance -> {
instance.runOnInstance(() -> {
backdateTimestamps(ttl + 2);
});
});
cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, ck, v) VALUES (400, 2, 2) IF NOT EXISTS", ConsistencyLevel.QUORUM);
// expire the cache entries
int nowInSec = FBUtilities.nowInSeconds();
cluster.get(1).runOnInstance(() -> {
TableMetadata table = Schema.instance.getTableMetadata(KEYSPACE, TABLE);
DecoratedKey dk = table.partitioner.decorateKey(ByteBufferUtil.bytes(400));
try (PaxosState state = PaxosState.get(dk, table))
{
state.updateStateUnsafe(s -> {
Assert.assertNull(s.accepted);
Assert.assertTrue(Commit.isAfter(s.committed.ballot, oldBallot));
Commit.CommittedWithTTL committed = new Commit.CommittedWithTTL(s.committed.ballot,
s.committed.update,
ballotDeletion(s.committed));
Assert.assertTrue(committed.localDeletionTime < nowInSec);
return new PaxosState.Snapshot(Ballot.none(), Ballot.none(), null, committed);
// prepare an operation ttl + 1 hour into the past on a single node
cluster.forEach(instance -> {
instance.runOnInstance(() -> {
Assert.assertEquals(OFFSETTABLE_CLOCK_NAME, CassandraRelevantProperties.CLOCK_GLOBAL.getString());
backdateTimestamps(ttl + 3600);
});
}
});
cluster.get(3).runOnInstance(() -> {
TableMetadata table = Schema.instance.getTableMetadata(KEYSPACE, TABLE);
DecoratedKey dk = table.partitioner.decorateKey(ByteBufferUtil.bytes(400));
try (PaxosState state = PaxosState.get(dk, table))
{
state.updateStateUnsafe(s -> {
Assert.assertNull(s.accepted);
Assert.assertTrue(Commit.isAfter(s.committed.ballot, oldBallot));
Commit.CommittedWithTTL committed = new Commit.CommittedWithTTL(s.committed.ballot,
s.committed.update,
ballotDeletion(s.committed));
Assert.assertTrue(committed.localDeletionTime < nowInSec);
return new PaxosState.Snapshot(oldBallot, oldBallot, null, committed);
});
}
});
cluster.forEach(instance -> {
instance.runOnInstance(() -> {
backdateTimestamps(0);
});
});
cluster.filters().inbound().to(1, 2).drop();
assertTimeout(() -> cluster.coordinator(3).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, ck, v) VALUES (400, 2, 2) IF NOT EXISTS", ConsistencyLevel.QUORUM));
Ballot oldBallot = Ballot.fromUuid(cluster.get(3).callOnInstance(() -> {
TableMetadata cfm = Schema.instance.getTableMetadata(KEYSPACE, TABLE);
DecoratedKey dk = cfm.partitioner.decorateKey(ByteBufferUtil.bytes(400));
try (PaxosState state = PaxosState.get(dk, cfm))
{
return state.currentSnapshot().promised.asUUID();
}
}));
cluster.filters().reset();
cluster.filters().inbound().to(2).drop();
cluster.get(3).runOnInstance(() -> {
assertUncommitted(cluster.get(1), KEYSPACE, TABLE, 0);
assertUncommitted(cluster.get(2), KEYSPACE, TABLE, 0);
assertUncommitted(cluster.get(3), KEYSPACE, TABLE, 1);
TableMetadata table = Schema.instance.getTableMetadata(KEYSPACE, TABLE);
DecoratedKey dk = table.partitioner.decorateKey(ByteBufferUtil.bytes(400));
// commit an operation just over ttl in the past on the other nodes
cluster.filters().reset();
cluster.filters().inbound().to(2).drop();
cluster.forEach(instance -> {
instance.runOnInstance(() -> {
backdateTimestamps(ttl + 2);
});
});
cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, ck, v) VALUES (400, 2, 2) IF NOT EXISTS", ConsistencyLevel.QUORUM);
UpdateSupplier supplier = PaxosState.uncommittedTracker().unsafGetUpdateSupplier();
try
{
PaxosUncommittedTracker.unsafSetUpdateSupplier(new SingleUpdateSupplier(table, dk, oldBallot));
StorageService.instance.autoRepairPaxos(table.id).get();
}
catch (Exception e)
{
throw new RuntimeException(e);
}
finally
{
PaxosUncommittedTracker.unsafSetUpdateSupplier(supplier);
}
});
// expire the cache entries
int nowInSec = FBUtilities.nowInSeconds();
cluster.get(1).runOnInstance(() -> {
TableMetadata table = Schema.instance.getTableMetadata(KEYSPACE, TABLE);
DecoratedKey dk = table.partitioner.decorateKey(ByteBufferUtil.bytes(400));
try (PaxosState state = PaxosState.get(dk, table))
{
state.updateStateUnsafe(s -> {
Assert.assertNull(s.accepted);
Assert.assertTrue(Commit.isAfter(s.committed.ballot, oldBallot));
Commit.CommittedWithTTL committed = new Commit.CommittedWithTTL(s.committed.ballot,
s.committed.update,
ballotDeletion(s.committed));
Assert.assertTrue(committed.localDeletionTime < nowInSec);
return new PaxosState.Snapshot(Ballot.none(), Ballot.none(), null, committed);
});
}
});
assertUncommitted(cluster.get(1), KEYSPACE, TABLE, 0);
assertUncommitted(cluster.get(2), KEYSPACE, TABLE, 0);
assertUncommitted(cluster.get(3), KEYSPACE, TABLE, 0);
cluster.get(3).runOnInstance(() -> {
TableMetadata table = Schema.instance.getTableMetadata(KEYSPACE, TABLE);
DecoratedKey dk = table.partitioner.decorateKey(ByteBufferUtil.bytes(400));
try (PaxosState state = PaxosState.get(dk, table))
{
state.updateStateUnsafe(s -> {
Assert.assertNull(s.accepted);
Assert.assertTrue(Commit.isAfter(s.committed.ballot, oldBallot));
Commit.CommittedWithTTL committed = new Commit.CommittedWithTTL(s.committed.ballot,
s.committed.update,
ballotDeletion(s.committed));
Assert.assertTrue(committed.localDeletionTime < nowInSec);
return new PaxosState.Snapshot(oldBallot, oldBallot, null, committed);
});
}
});
cluster.forEach(instance -> {
instance.runOnInstance(() -> {
backdateTimestamps(0);
});
});
cluster.filters().reset();
cluster.filters().inbound().to(2).drop();
cluster.get(3).runOnInstance(() -> {
TableMetadata table = Schema.instance.getTableMetadata(KEYSPACE, TABLE);
DecoratedKey dk = table.partitioner.decorateKey(ByteBufferUtil.bytes(400));
UpdateSupplier supplier = PaxosState.uncommittedTracker().unsafGetUpdateSupplier();
try
{
PaxosUncommittedTracker.unsafSetUpdateSupplier(new SingleUpdateSupplier(table, dk, oldBallot));
StorageService.instance.autoRepairPaxos(table.id).get();
}
catch (Exception e)
{
throw new RuntimeException(e);
}
finally
{
PaxosUncommittedTracker.unsafSetUpdateSupplier(supplier);
}
});
assertUncommitted(cluster.get(1), KEYSPACE, TABLE, 0);
assertUncommitted(cluster.get(2), KEYSPACE, TABLE, 0);
assertUncommitted(cluster.get(3), KEYSPACE, TABLE, 0);
}
}
finally
{
CassandraRelevantProperties.CLOCK_GLOBAL.reset();
}
}