diff --git a/CHANGES.txt b/CHANGES.txt index 210a834af5..9d2c9da3c1 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index bcf4dc7073..99ee013024 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -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); } /** diff --git a/src/java/org/apache/cassandra/db/Keyspace.java b/src/java/org/apache/cassandra/db/Keyspace.java index a06c7e137a..05b354a74b 100644 --- a/src/java/org/apache/cassandra/db/Keyspace.java +++ b/src/java/org/apache/cassandra/db/Keyspace.java @@ -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); diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index b8ecc5afb4..7bfbda4705 100644 --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java @@ -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; diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java index fc842db925..589cf39c77 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java @@ -636,6 +636,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte private class PaxosPurger extends Transformation { + private final long nowInSec; private final long paxosPurgeGraceMicros = DatabaseDescriptor.getPaxosPurgeGrace(MICROSECONDS); private final Map tableIdToHistory = new HashMap<>(); diff --git a/src/java/org/apache/cassandra/service/paxos/Commit.java b/src/java/org/apache/cassandra/service/paxos/Commit.java index 83402445a8..3aa8d65bce 100644 --- a/src/java/org/apache/cassandra/service/paxos/Commit.java +++ b/src/java/org/apache/cassandra/service/paxos/Commit.java @@ -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 diff --git a/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java b/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java index a7f58e1383..c0a2353e1f 100644 --- a/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java +++ b/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java @@ -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 im private final List> readResponses; private boolean haveReadResponseWithLatest; private boolean haveQuorumOfPermissions; // permissions => SUCCESS or READ_SUCCESS - private List withLatest; // promised and have latest commit - private List needLatest; // promised without having witnessed latest commit, nor yet been refreshed by us + private @Nonnull List withLatest; // promised and have latest commit + private @Nullable List 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 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(); diff --git a/src/java/org/apache/cassandra/service/paxos/uncommitted/PaxosRows.java b/src/java/org/apache/cassandra/service/paxos/uncommitted/PaxosRows.java index ee5c762a1f..7990a80e32 100644 --- a/src/java/org/apache/cassandra/service/paxos/uncommitted/PaxosRows.java +++ b/src/java/org/apache/cassandra/service/paxos/uncommitted/PaxosRows.java @@ -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) diff --git a/test/distributed/org/apache/cassandra/distributed/test/CasWriteTest.java b/test/distributed/org/apache/cassandra/distributed/test/CasWriteTest.java index bb4f535286..d6bc4a39b2 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/CasWriteTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/CasWriteTest.java @@ -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 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; + } } diff --git a/test/unit/org/apache/cassandra/service/paxos/PaxosStateTest.java b/test/unit/org/apache/cassandra/service/paxos/PaxosStateTest.java index 893b1d2f0c..3778044233 100644 --- a/test/unit/org/apache/cassandra/service/paxos/PaxosStateTest.java +++ b/test/unit/org/apache/cassandra/service/paxos/PaxosStateTest.java @@ -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 test) diff --git a/test/unit/org/apache/cassandra/streaming/StreamSessionTest.java b/test/unit/org/apache/cassandra/streaming/StreamSessionTest.java index 44491d55e5..4073a466a3 100644 --- a/test/unit/org/apache/cassandra/streaming/StreamSessionTest.java +++ b/test/unit/org/apache/cassandra/streaming/StreamSessionTest.java @@ -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());