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

Refresh stale paxos commit 

patch by Benedict Elliott Smith; review by Blake Eggleston for CASSANDRA-19617
This commit is contained in:
Benedict Elliott Smith 2024-04-26 12:50:40 +01:00 committed by Benjamin Lerer
parent 786ba38200
commit 8c165b4cc0
11 changed files with 262 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.0-beta2
* Refresh stale paxos commit (CASSANDRA-19617)
* Replace Stream iteration with for-loop for StorageProxy::updateCoordinatorWriteLatencyTableMetric (CASSANDRA-19676)
* Enforce metric naming contract if scope is used in a metric name (CASSANDRA-19619)
* Avoid reading of the same IndexInfo from disk many times for a large partition (CASSANDRA-19557)

15
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -3306,13 +3306,16 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
if (metadata == null)
return null;
Keyspace keyspace = Keyspace.open(metadata.keyspace);
if (keyspace == null)
return null;
return getIfExists(metadata);
}
return keyspace.hasColumnFamilyStore(id)
? keyspace.getColumnFamilyStore(id)
: null;
/**
* Returns a ColumnFamilyStore by metadata if it exists, null otherwise
* Differently from others, this method does not throw exception if the table does not exist.
*/
public static ColumnFamilyStore getIfExists(TableMetadata table)
{
return Keyspace.openAndGetStoreIfExists(table);
}
/**

13
src/java/org/apache/cassandra/db/Keyspace.java
View File

@ -172,6 +172,14 @@ public class Keyspace
return open(table.keyspace).getColumnFamilyStore(table.id);
}
public static ColumnFamilyStore openAndGetStoreIfExists(TableMetadata table)
{
Keyspace keyspace = open(table.keyspace);
if (keyspace == null)
return null;
return keyspace.getIfExists(table.id);
}
/**
* Removes every SSTable in the directory from the appropriate Tracker's view.
* @param directory the unreadable directory, possibly with SSTables in it, but not necessarily.
@ -220,6 +228,11 @@ public class Keyspace
return cfs;
}
public ColumnFamilyStore getIfExists(TableId id)
{
return columnFamilyStores.get(id);
}
public boolean hasColumnFamilyStore(TableId id)
{
return columnFamilyStores.containsKey(id);

31
src/java/org/apache/cassandra/db/SystemKeyspace.java
View File

@ -122,6 +122,7 @@ import org.apache.cassandra.utils.concurrent.Future;
import static java.lang.String.format;
import static java.util.Collections.emptyMap;
import static java.util.Collections.singletonMap;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
import static org.apache.cassandra.config.Config.PaxosStatePurging.legacy;
import static org.apache.cassandra.config.DatabaseDescriptor.paxosStatePurging;
import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
@ -1318,6 +1319,8 @@ public final class SystemKeyspace
/**
* Load the current paxos state for the table and key
*
* NOTE: nowInSec is typically provided as zero, and should not be assumed to be definitive, as the cache may apply different nowInSec filters
*/
public static PaxosState.Snapshot loadPaxosState(DecoratedKey partitionKey, TableMetadata metadata, long nowInSec)
{
@ -1329,14 +1332,38 @@ public final class SystemKeyspace
return new PaxosState.Snapshot(Ballot.none(), Ballot.none(), null, noneCommitted);
}
long purgeBefore = 0;
long overrideTtlSeconds = 0;
switch (paxosStatePurging())
{
default: throw new AssertionError();
case legacy:
case gc_grace:
overrideTtlSeconds = metadata.params.gcGraceSeconds;
if (nowInSec > 0)
purgeBefore = TimeUnit.SECONDS.toMicros(nowInSec - overrideTtlSeconds);
break;
case repaired:
ColumnFamilyStore cfs = Keyspace.openAndGetStoreIfExists(metadata);
if (cfs != null)
{
long paxosPurgeGraceMicros = DatabaseDescriptor.getPaxosPurgeGrace(MICROSECONDS);
purgeBefore = cfs.getPaxosRepairLowBound(partitionKey).uuidTimestamp() - paxosPurgeGraceMicros;
}
}
Row row = results.get(0);
Ballot promisedWrite = PaxosRows.getWritePromise(row);
if (promisedWrite.uuidTimestamp() < purgeBefore) promisedWrite = Ballot.none();
Ballot promised = latest(promisedWrite, PaxosRows.getPromise(row));
if (promised.uuidTimestamp() < purgeBefore) promised = Ballot.none();
// either we have both a recently accepted ballot and update or we have neither
Accepted accepted = PaxosRows.getAccepted(row);
Committed committed = PaxosRows.getCommitted(metadata, partitionKey, row);
Accepted accepted = PaxosRows.getAccepted(row, purgeBefore, overrideTtlSeconds);
Committed committed = PaxosRows.getCommitted(metadata, partitionKey, row, purgeBefore, overrideTtlSeconds);
// fix a race with TTL/deletion resolution, where TTL expires after equal deletion is inserted; TTL wins the resolution, and is read using an old ballot's nowInSec
if (accepted != null && !accepted.isAfter(committed))
accepted = null;

1
src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
View File

@ -636,6 +636,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
private class PaxosPurger extends Transformation<UnfilteredRowIterator>
{
private final long nowInSec;
private final long paxosPurgeGraceMicros = DatabaseDescriptor.getPaxosPurgeGrace(MICROSECONDS);
private final Map<TableId, PaxosRepairHistory.Searcher> tableIdToHistory = new HashMap<>();

5
src/java/org/apache/cassandra/service/paxos/Commit.java
View File

@ -212,6 +212,11 @@ public class Commit
return c > 0 ? a : b;
return a instanceof CommittedWithTTL ? ((CommittedWithTTL)a).lastDeleted(b) : a;
}
public boolean isNone()
{
return ballot.equals(Ballot.none()) && update.isEmpty();
}
}
public static class CommittedWithTTL extends Committed

22
src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java
View File

@ -27,6 +27,7 @@ import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import org.slf4j.Logger;
@ -286,8 +287,8 @@ public class PaxosPrepare extends PaxosRequestCallback<PaxosPrepare.Response> im
private final List<Message<ReadResponse>> readResponses;
private boolean haveReadResponseWithLatest;
private boolean haveQuorumOfPermissions; // permissions => SUCCESS or READ_SUCCESS
private List<InetAddressAndPort> withLatest; // promised and have latest commit
private List<InetAddressAndPort> needLatest; // promised without having witnessed latest commit, nor yet been refreshed by us
private @Nonnull List<InetAddressAndPort> withLatest; // promised and have latest commit
private @Nullable List<InetAddressAndPort> needLatest; // promised without having witnessed latest commit, nor yet been refreshed by us
private int failures; // failed either on initial request or on refresh
private boolean hasProposalStability = true; // no successful modifying proposal could have raced with us and not been seen
private boolean hasOnlyPromises = true;
@ -505,11 +506,26 @@ public class PaxosPrepare extends PaxosRequestCallback<PaxosPrepare.Response> im
}
if (permitted.lowBound > maxLowBound)
{
maxLowBound = permitted.lowBound;
if (!latestCommitted.isNone() && latestCommitted.ballot.uuidTimestamp() < maxLowBound)
{
latestCommitted = Committed.none(request.partitionKey, request.table);
haveReadResponseWithLatest = !readResponses.isEmpty();
if (needLatest != null)
{
withLatest.addAll(needLatest);
needLatest.clear();
}
}
}
if (!haveQuorumOfPermissions)
{
CompareResult compareLatest = permitted.latestCommitted.compareWith(latestCommitted);
Committed newLatestCommitted = permitted.latestCommitted;
if (newLatestCommitted.ballot.uuidTimestamp() < maxLowBound) newLatestCommitted = Committed.none(request.partitionKey, request.table);
CompareResult compareLatest = newLatestCommitted.compareWith(latestCommitted);
switch (compareLatest)
{
default: throw new IllegalStateException();

23
src/java/org/apache/cassandra/service/paxos/uncommitted/PaxosRows.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.service.paxos.uncommitted;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
@ -88,32 +89,38 @@ public class PaxosRows
return getBallot(row, WRITE_PROMISE, Ballot.none());
}
public static Accepted getAccepted(Row row)
public static Accepted getAccepted(Row row, long purgeBefore, long overrideTtlSeconds)
{
Cell ballotCell = row.getCell(PROPOSAL);
if (ballotCell == null)
return null;
Ballot ballot = ballotCell.accessor().toBallot(ballotCell.value());
if (ballot.uuidTimestamp() < purgeBefore)
return null;
int version = getInt(row, PROPOSAL_VERSION, MessagingService.VERSION_40);
PartitionUpdate update = getUpdate(row, PROPOSAL_UPDATE, version);
return ballotCell.isExpiring()
? new AcceptedWithTTL(ballot, update, ballotCell.localDeletionTime())
: new Accepted(ballot, update);
if (overrideTtlSeconds > 0) return new AcceptedWithTTL(ballot, update, TimeUnit.MICROSECONDS.toSeconds(ballotCell.timestamp()) + overrideTtlSeconds);
else if (ballotCell.isExpiring()) return new AcceptedWithTTL(ballot, update, ballotCell.localDeletionTime());
else return new Accepted(ballot, update);
}
public static Committed getCommitted(TableMetadata metadata, DecoratedKey partitionKey, Row row)
public static Committed getCommitted(TableMetadata metadata, DecoratedKey partitionKey, Row row, long purgeBefore, long overrideTtlSeconds)
{
Cell ballotCell = row.getCell(COMMIT);
if (ballotCell == null)
return Committed.none(partitionKey, metadata);
Ballot ballot = ballotCell.accessor().toBallot(ballotCell.value());
if (ballot.uuidTimestamp() < purgeBefore)
return Committed.none(partitionKey, metadata);
int version = getInt(row, COMMIT_VERSION, MessagingService.VERSION_40);
PartitionUpdate update = getUpdate(row, COMMIT_UPDATE, version);
return ballotCell.isExpiring()
? new CommittedWithTTL(ballot, update, ballotCell.localDeletionTime())
: new Committed(ballot, update);
if (overrideTtlSeconds > 0) return new CommittedWithTTL(ballot, update, TimeUnit.MICROSECONDS.toSeconds(ballotCell.timestamp()) + overrideTtlSeconds);
else if (ballotCell.isExpiring()) return new CommittedWithTTL(ballot, update, ballotCell.localDeletionTime());
else return new Committed(ballot, update);
}
public static TableId getTableId(Row row)

147
test/distributed/org/apache/cassandra/distributed/test/CasWriteTest.java
View File

@ -44,19 +44,33 @@ import org.junit.rules.ExpectedException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.SinglePartitionReadCommand;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.ICluster;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.IMessageFilters;
import org.apache.cassandra.distributed.shared.InstanceClassLoader;
import org.apache.cassandra.exceptions.CasWriteTimeoutException;
import org.apache.cassandra.exceptions.CasWriteUnknownResultException;
import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.notifications.SSTableMetadataChanged;
import org.apache.cassandra.service.paxos.Ballot;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.TimeUUID;
import org.hamcrest.BaseMatcher;
import org.hamcrest.Description;
import static org.apache.cassandra.db.ColumnFamilyStore.FlushReason.INTERNALLY_FORCED;
import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows;
import static org.apache.cassandra.distributed.shared.AssertUtils.row;
import static org.hamcrest.CoreMatchers.containsString;
@ -68,6 +82,7 @@ public class CasWriteTest extends TestBaseImpl
private static ICluster cluster;
private static final AtomicInteger pkGen = new AtomicInteger(1_000); // preserve any pk values less than 1000 for manual queries.
private static final Logger logger = LoggerFactory.getLogger(CasWriteTest.class);
private static final long GC_GRACE_SECONDS = 10;
@Rule
public ExpectedException thrown = ExpectedException.none();
@ -75,8 +90,10 @@ public class CasWriteTest extends TestBaseImpl
@BeforeClass
public static void setupCluster() throws Throwable
{
cluster = init(Cluster.build().withNodes(3).start());
cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
cluster = init(Cluster.build().withNodes(3).withConfig(config -> config.set("paxos_state_purging", "repaired")
.set("paxos_variant", "v2")
.set("paxos_cache_size", "0MiB")).start());
cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck)) WITH gc_grace_seconds = " + GC_GRACE_SECONDS);
}
@AfterClass
@ -295,6 +312,125 @@ public class CasWriteTest extends TestBaseImpl
Assert.fail("Expecting test to throw a CasWriteUnknownResultException");
}
@Test
public void testStaleCommitInSystemPaxos() throws InterruptedException
{
cluster.filters().reset();
int extraKeys = 10;
int pk = pkGen.getAndAdd(extraKeys + 1);
cluster.coordinator(1).execute(mkCasInsertQuery((a) -> pk, 1, 1), ConsistencyLevel.ALL);
for (int i = 1 ; i <= 3 ; ++i)
{
((IInvokableInstance)cluster.get(i)).runOnInstance(() -> DatabaseDescriptor.setPaxosPurgeGrace(0));
}
long insertTimestamp = ((IInvokableInstance)cluster.get(3)).applyOnInstance(pk_ -> {
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl");
DecoratedKey key = cfs.decorateKey(Int32Type.instance.decompose(pk_));
return SystemKeyspace.loadPaxosState(key, cfs.metadata.get(), FBUtilities.nowInSeconds()).committed.ballot.uuidTimestamp();
}, pk);
((IInvokableInstance)cluster.get(3)).runOnInstance(() -> {
ColumnFamilyStore cfs = Keyspace.open("system").getColumnFamilyStore("paxos");
cfs.forceFlush(INTERNALLY_FORCED).awaitUninterruptibly();
cfs.getLiveSSTables().forEach(s -> {
try
{
StatsMetadata oldMetadata = s.getSSTableMetadata();
s.mutateLevelAndReload(3);
cfs.getCompactionStrategyManager().handleNotification(new SSTableMetadataChanged(s, oldMetadata), null);
}
catch (Throwable t)
{
t.printStackTrace();
}
});
});
cluster.coordinator(1).execute(mkCasDeleteQuery((a) -> pk, 1, 1), ConsistencyLevel.ALL);
long deleteTimestamp = ((IInvokableInstance)cluster.get(3)).applyOnInstance(pk_ -> {
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl");
DecoratedKey key = cfs.decorateKey(Int32Type.instance.decompose(pk_));
return SystemKeyspace.loadPaxosState(key, cfs.metadata.get(), FBUtilities.nowInSeconds()).committed.ballot.uuidTimestamp();
}, pk);
cluster.get(1).nodetool("repair", "--paxos-only", KEYSPACE, "tbl");
// write and flush enough data to trigger purge of the deletion commit, without touching the earlier insertion commit that is in a higher level
for (int i = 0 ; i < 10 ; ++i)
{
for (int j = 1; j <= extraKeys ; ++j)
{
final int pkj = pk + j;
cluster.coordinator(1).execute(mkCasInsertQuery(a -> pkj, i, 1), ConsistencyLevel.ALL);
}
for (int k = 1 ; k <= 3 ; ++k)
{
((IInvokableInstance)cluster.get(k)).runOnInstance(() -> {
ColumnFamilyStore cfs = Keyspace.open("system").getColumnFamilyStore("paxos");
cfs.forceFlush(INTERNALLY_FORCED).awaitUninterruptibly();
ColumnFamilyStore cfs2 = Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl");
cfs2.forceFlush(INTERNALLY_FORCED).awaitUninterruptibly();
});
}
}
for (int k = 1 ; k <= 3 ; ++k)
{
((IInvokableInstance)cluster.get(k)).runOnInstance(() -> {
ColumnFamilyStore cfs = Keyspace.open("system").getColumnFamilyStore("paxos");
while (cfs.getCompactionStrategyManager().getEstimatedRemainingTasks() > 0)
{
try { Thread.sleep(1000); }
catch (InterruptedException e) { throw new RuntimeException(e); }
}
});
}
long repairTimestamp = ((IInvokableInstance)cluster.get(3)).applyOnInstance(pk_ -> {
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl");
DecoratedKey key = cfs.decorateKey(Int32Type.instance.decompose(pk_));
return cfs.getPaxosRepairHistory().ballotForToken(key.getToken()).uuidTimestamp();
}, pk);
long afterRepairTimestampOn1 = ((IInvokableInstance)cluster.get(1)).applyOnInstance(pk_ -> {
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl");
DecoratedKey key = cfs.decorateKey(Int32Type.instance.decompose(pk_));
return SystemKeyspace.loadPaxosState(key, cfs.metadata.get(), FBUtilities.nowInSeconds()).committed.ballot.uuidTimestamp();
}, pk);
long afterRepairTimestampOn3 = ((IInvokableInstance)cluster.get(3)).applyOnInstance(pk_ -> {
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl");
DecoratedKey key = cfs.decorateKey(Int32Type.instance.decompose(pk_));
return SystemKeyspace.loadPaxosState(key, cfs.metadata.get(), FBUtilities.nowInSeconds()).committed.ballot.uuidTimestamp();
}, pk);
Assert.assertEquals(Ballot.none().uuidTimestamp(), afterRepairTimestampOn1);
logger.info("Waiting for tombstone to be purgeable");
Thread.sleep(GC_GRACE_SECONDS * 1000);
while (FBUtilities.timestampMicros() - (GC_GRACE_SECONDS * 1000_000) < TimeUUID.rawTimestampToUnixMicros(deleteTimestamp))
Thread.sleep(1000);
cluster.get(1).nodetool("compact", KEYSPACE, "tbl");
for (int i = 1 ; i <= 3 ; ++i)
{
int partitionCount = ((IInvokableInstance)cluster.get(3)).applyOnInstance(pk_ -> {
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl");
DecoratedKey key = cfs.decorateKey(Int32Type.instance.decompose(pk_));
return Util.getAllUnfiltered(SinglePartitionReadCommand.create(cfs.metadata.get(), FBUtilities.nowInSeconds(), key, cfs.metadata.get().comparator.make(Int32Type.instance.decompose(1)))).size();
}, pk);
Assert.assertEquals(0, partitionCount);
}
cluster.filters().allVerbs().from(1).to(2).drop();
// we must first perform a write as the read has proposal stability and so responds async
cluster.coordinator(1).execute("UPDATE " + KEYSPACE + ".tbl SET v = 2 WHERE ck = 1 AND pk = " + pk + " IF EXISTS", ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM);
Assert.assertArrayEquals(new Object[0], cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = " + pk, ConsistencyLevel.SERIAL));
}
private static boolean isPaxosVariant2()
{
return Config.PaxosVariant.v2.name().equals(cluster.coordinator(1).instance().config().getString("paxos_variant"));
@ -312,4 +448,11 @@ public class CasWriteTest extends TestBaseImpl
logger.info("Generated query: " + query);
return query;
}
private String mkCasDeleteQuery(Function<AtomicInteger, Integer> pkFunc, int ck, int v)
{
String query = String.format("DELETE FROM %s.tbl WHERE pk = %d AND ck = 1 IF EXISTS", KEYSPACE, pkFunc.apply(pkGen));
logger.info("Generated query: " + query);
return query;
}
}

45
test/unit/org/apache/cassandra/service/paxos/PaxosStateTest.java
View File

@ -40,6 +40,7 @@ import org.apache.cassandra.service.paxos.PaxosState.Snapshot;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
import static org.apache.cassandra.config.Config.PaxosStatePurging.gc_grace;
import static org.apache.cassandra.config.Config.PaxosStatePurging.legacy;
import static org.apache.cassandra.config.Config.PaxosStatePurging.repaired;
@ -104,34 +105,36 @@ public class PaxosStateTest
String key = "key" + System.nanoTime();
Accepted accepted = newProposal(1, key).accepted();
AcceptedWithTTL acceptedWithTTL = new AcceptedWithTTL(accepted, (int)MICROSECONDS.toSeconds(accepted.ballot.unixMicros())-3600);
PaxosState.legacyPropose(accepted);
DatabaseDescriptor.setPaxosStatePurging(repaired);
// not expired if read in the past
assertPaxosState(key, accepted, state -> state.current(accepted.ballot).accepted);
// not expired if read with paxos state purging enabled
assertPaxosState(key, accepted, state -> state.current(Long.MAX_VALUE).accepted);
assertPaxosState(key, accepted, state -> state.current(Integer.MAX_VALUE).accepted);
DatabaseDescriptor.setPaxosStatePurging(gc_grace);
// not expired if read in the past
assertPaxosState(key, accepted, state -> state.current(accepted.ballot).accepted);
assertPaxosState(key, acceptedWithTTL, state -> state.current(accepted.ballot).accepted);
// expired if read with paxos state purging disabled
assertPaxosState(key, null, state -> state.current(Long.MAX_VALUE).accepted);
assertPaxosState(key, null, state -> state.current(Integer.MAX_VALUE).accepted);
// clear cache to read from disk
PaxosState.RECENT.clear();
Committed committed = accepted.committed();
CommittedWithTTL committedWithTTL = new CommittedWithTTL(committed, (int)MICROSECONDS.toSeconds(committed.ballot.unixMicros())-3600);
Committed empty = emptyProposal(key).accepted().committed();
PaxosState.commitDirect(committed);
DatabaseDescriptor.setPaxosStatePurging(repaired);
// not expired if read in the past
assertPaxosState(key, committed, state -> state.current(committed.ballot).committed);
// not expired if read with paxos state purging enabled
assertPaxosState(key, committed, state -> state.current(Long.MAX_VALUE).committed);
assertPaxosState(key, committed, state -> state.current(Integer.MAX_VALUE).committed);
DatabaseDescriptor.setPaxosStatePurging(gc_grace);
// not expired if read in the past
assertPaxosState(key, committed, state -> state.current(committed.ballot).committed);
assertPaxosState(key, committedWithTTL, state -> state.current(committed.ballot).committed);
// expired if read with paxos state purging disabled
assertPaxosState(key, empty, state -> state.current(Long.MAX_VALUE).committed);
assertPaxosState(key, empty, state -> state.current(Integer.MAX_VALUE).committed);
DatabaseDescriptor.setPaxosStatePurging(repaired);
}
@ -140,7 +143,7 @@ public class PaxosStateTest
{
String key = "key" + System.nanoTime();
String key2 = key + 'A';
Accepted accepted = new AcceptedWithTTL(newProposal(1, key), 1);
Accepted accepted = new AcceptedWithTTL(newProposal(1, key), SystemKeyspace.legacyPaxosTtlSec(metadata) + 1);
PaxosState.legacyPropose(accepted);
PaxosState.legacyPropose(new AcceptedWithTTL(newProposal(1, key2), 10000));
@ -148,21 +151,21 @@ public class PaxosStateTest
// not expired if read in the past
assertPaxosState(key, accepted, state -> state.current(0).accepted);
// TTL, so still expired if read with paxos state purging enabled
assertPaxosState(key, null, state -> state.current(Long.MAX_VALUE).accepted);
assertPaxosState(key, null, state -> state.current(Integer.MAX_VALUE).accepted);
DatabaseDescriptor.setPaxosStatePurging(gc_grace);
// not expired if read in the past
assertPaxosState(key, accepted, state -> state.current(0).accepted);
// expired if read with paxos state purging disabled
assertPaxosState(key, null, state -> state.current(Long.MAX_VALUE).accepted);
assertPaxosState(key, null, state -> state.current(Integer.MAX_VALUE).accepted);
DatabaseDescriptor.setPaxosStatePurging(legacy);
// not expired if read in the past
assertPaxosState(key, accepted, state -> state.current(0).accepted);
// expired if read with paxos state purging disabled
assertPaxosState(key, null, state -> state.current(Long.MAX_VALUE).accepted);
assertPaxosState(key, null, state -> state.current(Integer.MAX_VALUE).accepted);
// clear cache to read from disk
PaxosState.RECENT.clear();
Committed committed = new CommittedWithTTL(accepted, accepted.update.metadata().params.gcGraceSeconds + 1);
Committed committed = new CommittedWithTTL(accepted, SystemKeyspace.legacyPaxosTtlSec(metadata) + 1);
Committed empty = emptyProposal(key).accepted().committed();
PaxosState.commitDirect(committed);
@ -170,17 +173,17 @@ public class PaxosStateTest
// not expired if read in the past
assertPaxosState(key, committed, state -> state.current(0).committed);
// not expired if read with paxos state purging enabled
assertPaxosState(key, empty, state -> state.current(Long.MAX_VALUE).committed);
assertPaxosState(key, empty, state -> state.current(Integer.MAX_VALUE).committed);
DatabaseDescriptor.setPaxosStatePurging(gc_grace);
// not expired if read in the past
assertPaxosState(key, committed, state -> state.current(0).committed);
// expired if read with paxos state purging disabled
assertPaxosState(key, empty, state -> state.current(Long.MAX_VALUE).committed);
assertPaxosState(key, empty, state -> state.current(Integer.MAX_VALUE).committed);
DatabaseDescriptor.setPaxosStatePurging(legacy);
// not expired if read in the past
assertPaxosState(key, committed, state -> state.current(0).committed);
// expired if read with paxos state purging disabled
assertPaxosState(key, empty, state -> state.current(Long.MAX_VALUE).committed);
assertPaxosState(key, empty, state -> state.current(Integer.MAX_VALUE).committed);
}
@Test
@ -197,43 +200,43 @@ public class PaxosStateTest
// not expired if read in the past (or now)
assertPaxosState(key, accepted, state -> state.current(0).accepted);
// TTL, so still expired if read with paxos state purging enabled
assertPaxosState(key, null, state -> state.current(Long.MAX_VALUE).accepted);
assertPaxosState(key, null, state -> state.current(Integer.MAX_VALUE).accepted);
DatabaseDescriptor.setPaxosStatePurging(gc_grace);
// not expired if read in the past (or now)
assertPaxosState(key, accepted, state -> state.current(0).accepted);
// expired if read with paxos state purging disabled
assertPaxosState(key, null, state -> state.current(Long.MAX_VALUE).accepted);
assertPaxosState(key, null, state -> state.current(Integer.MAX_VALUE).accepted);
DatabaseDescriptor.setPaxosStatePurging(legacy);
// not expired if read in the past (or now)
assertPaxosState(key, accepted, state -> state.current(0).accepted);
// TTL, so expired in the future
assertPaxosState(key, null, state -> state.current(Long.MAX_VALUE).accepted);
assertPaxosState(key, null, state -> state.current(Integer.MAX_VALUE).accepted);
PaxosState.RECENT.clear();
Committed committed = new Committed(accepted);
Committed empty = emptyProposal(key).accepted().committed();
DatabaseDescriptor.setPaxosStatePurging(legacy); // write with TTLs
committed = new CommittedWithTTL(committed, committed.update.metadata().params.gcGraceSeconds + 1); // for equality test
PaxosState.commitDirect(committed);
committed = new CommittedWithTTL(committed, -1); // for equality test
DatabaseDescriptor.setPaxosStatePurging(repaired);
// not expired if read in the past
assertPaxosState(key, committed, state -> state.current(0).committed);
// not expired if read with paxos state purging enabled
assertPaxosState(key, empty, state -> state.current(Long.MAX_VALUE).committed);
assertPaxosState(key, empty, state -> state.current(Integer.MAX_VALUE).committed);
DatabaseDescriptor.setPaxosStatePurging(gc_grace);
// not expired if read in the past
assertPaxosState(key, committed, state -> state.current(0).committed);
// expired if read with paxos state purging disabled
assertPaxosState(key, empty, state -> state.current(Long.MAX_VALUE).committed);
assertPaxosState(key, empty, state -> state.current(Integer.MAX_VALUE).committed);
DatabaseDescriptor.setPaxosStatePurging(legacy);
// not expired if read in the past
assertPaxosState(key, committed, state -> state.current(0).committed);
// expired if read with paxos state purging disabled
assertPaxosState(key, empty, state -> state.current(Long.MAX_VALUE).committed);
assertPaxosState(key, empty, state -> state.current(Integer.MAX_VALUE).committed);
}
private static void assertPaxosState(String key, Commit expect, Function<PaxosState, Object> test)

2
test/unit/org/apache/cassandra/streaming/StreamSessionTest.java
View File

@ -72,7 +72,7 @@ public class StreamSessionTest extends CQLTester
DatabaseDescriptor.daemonInitialization();
ByteBuddyAgent.install();
new ByteBuddy().redefine(ColumnFamilyStore.class)
.method(named("getIfExists").and(takesArguments(1)))
.method(named("getIfExists").and(takesArguments(TableId.class)))
.intercept(MethodDelegation.to(BBKeyspaceHelper.class))
.make()
.load(ColumnFamilyStore.class.getClassLoader(), ClassReloadingStrategy.fromInstalledAgent());