You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by aw...@apache.org on 2024/05/29 22:00:54 UTC
(cassandra) branch cep-15-accord updated (aa55d9ed30 -> 3b99044d6d)
This is an automated email from the ASF dual-hosted git repository.
aweisberg pushed a change to branch cep-15-accord
in repository https://gitbox.apache.org/repos/asf/cassandra.git
discard aa55d9ed30 Accord barrier/inclusive sync point fixes
new 3b99044d6d Accord barrier/inclusive sync point fixes
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (aa55d9ed30)
\
N -- N -- N refs/heads/cep-15-accord (3b99044d6d)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.gitmodules | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 01/01: Accord barrier/inclusive sync point fixes
Posted by aw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aweisberg pushed a commit to branch cep-15-accord
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 3b99044d6d5491304d4a25d8dcea54510cfd3215
Author: Ariel Weisberg <aw...@apple.com>
AuthorDate: Fri May 17 15:27:44 2024 -0400
Accord barrier/inclusive sync point fixes
Patch by Ariel Weisberg, Benedict Elliott Smith; reviewed by Benedict Elliott Smith for CASSANDRA-19641
---
modules/accord | 2 +-
.../cassandra/service/accord/AccordService.java | 2 +-
.../cassandra/service/accord/api/AccordAgent.java | 10 +-
.../migration/ConsensusKeyMigrationState.java | 20 ++-
.../test/accord/AccordIncrementalRepairTest.java | 6 +-
.../test/accord/AccordMigrationTest.java | 169 ++++++++++++++-------
6 files changed, 132 insertions(+), 77 deletions(-)
diff --git a/modules/accord b/modules/accord
index 778c45cd97..4e8bcae81f 160000
--- a/modules/accord
+++ b/modules/accord
@@ -1 +1 @@
-Subproject commit 778c45cd977576a901abf24a9759872d36fde056
+Subproject commit 4e8bcae81f9751b9d732fd5056bce31c97ad58f3
diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java
index f853c329c6..8e2778f394 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordService.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordService.java
@@ -366,7 +366,7 @@ public class AccordService implements IAccordService, Shutdownable
{
logger.debug("Starting barrier key: {} epoch: {} barrierType: {} isForWrite {}", keysOrRanges, epoch, barrierType, isForWrite);
txnId = node.nextTxnId(Kind.SyncPoint, keysOrRanges.domain());
- AsyncResult<Timestamp> asyncResult = syncPoint == null
+ AsyncResult<TxnId> asyncResult = syncPoint == null
? Barrier.barrier(node, keysOrRanges, epoch, barrierType)
: Barrier.barrier(node, keysOrRanges, epoch, barrierType, syncPoint);
long deadlineNanos = queryStartNanos + timeoutNanos;
diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
index 9c4b678996..c0fea38a37 100644
--- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
+++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
@@ -19,12 +19,11 @@
package org.apache.cassandra.service.accord.api;
import java.util.concurrent.TimeUnit;
+import javax.annotation.Nonnull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.annotation.Nonnull;
-
import accord.api.Agent;
import accord.api.EventsListener;
import accord.api.Result;
@@ -35,8 +34,9 @@ import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.Txn.Kind;
-import org.apache.cassandra.service.accord.AccordService;
+import accord.primitives.TxnId;
import org.apache.cassandra.metrics.AccordMetrics;
+import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.txn.TxnQuery;
import org.apache.cassandra.service.accord.txn.TxnRead;
import org.apache.cassandra.tcm.Epoch;
@@ -84,12 +84,12 @@ public class AccordAgent implements Agent
}
@Override
- public void onLocalBarrier(@Nonnull Seekables<?, ?> keysOrRanges, @Nonnull Timestamp executeAt)
+ public void onLocalBarrier(@Nonnull Seekables<?, ?> keysOrRanges, @Nonnull TxnId txnId)
{
if (keysOrRanges.domain() == Key)
{
PartitionKey key = (PartitionKey)keysOrRanges.get(0);
- maybeSaveAccordKeyMigrationLocally(key, Epoch.create(executeAt.epoch()));
+ maybeSaveAccordKeyMigrationLocally(key, Epoch.create(txnId.epoch()));
}
}
diff --git a/src/java/org/apache/cassandra/service/consensus/migration/ConsensusKeyMigrationState.java b/src/java/org/apache/cassandra/service/consensus/migration/ConsensusKeyMigrationState.java
index 068e46d4a4..86eb97ae06 100644
--- a/src/java/org/apache/cassandra/service/consensus/migration/ConsensusKeyMigrationState.java
+++ b/src/java/org/apache/cassandra/service/consensus/migration/ConsensusKeyMigrationState.java
@@ -27,6 +27,8 @@ import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.primitives.Ints;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import accord.api.BarrierType;
import accord.primitives.Seekables;
@@ -68,9 +70,7 @@ import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.UUIDSerializer;
import static org.apache.cassandra.net.Verb.CONSENSUS_KEY_MIGRATION;
-
import static org.apache.cassandra.service.consensus.migration.ConsensusMigrationTarget.paxos;
-
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
@@ -83,6 +83,8 @@ import static org.apache.cassandra.utils.Clock.Global.nanoTime;
*/
public abstract class ConsensusKeyMigrationState
{
+ private static final Logger logger = LoggerFactory.getLogger(ConsensusKeyMigrationState.class);
+
/*
* Used to notify other replicas when key migration has occurred so they can
* also cache that the key migration was done
@@ -188,12 +190,14 @@ public abstract class ConsensusKeyMigrationState
private static final CacheLoader<Pair<ByteBuffer, UUID>, ConsensusMigratedAt> LOADING_FUNCTION = k -> SystemKeyspace.loadConsensusKeyMigrationState(k.left, k.right);
private static final Weigher<Pair<ByteBuffer, UUID>, ConsensusMigratedAt> WEIGHER_FUNCTION = (k, v) -> EMPTY_KEY_SIZE + Ints.checkedCast(ByteBufferUtil.estimatedSizeOnHeap(k.left)) + VALUE_SIZE;
- private static final LoadingCache<Pair<ByteBuffer, UUID>, ConsensusMigratedAt> MIGRATION_STATE_CACHE =
- Caffeine.newBuilder()
- .maximumWeight(DatabaseDescriptor.getConsensusMigrationCacheSizeInMiB() << 20)
- .weigher(WEIGHER_FUNCTION)
- .executor(ImmediateExecutor.INSTANCE)
- .build(LOADING_FUNCTION);
+
+ @VisibleForTesting
+ public static final LoadingCache<Pair<ByteBuffer, UUID>, ConsensusMigratedAt> MIGRATION_STATE_CACHE =
+ Caffeine.newBuilder()
+ .maximumWeight(DatabaseDescriptor.getConsensusMigrationCacheSizeInMiB() << 20)
+ .weigher(WEIGHER_FUNCTION)
+ .executor(ImmediateExecutor.INSTANCE)
+ .build(LOADING_FUNCTION);
public static final IVerbHandler<ConsensusKeyMigrationFinished> consensusKeyMigrationFinishedHandler = message -> {
saveConsensusKeyMigrationLocally(message.payload.partitionKey, message.payload.tableId, message.payload.consensusMigratedAt);
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java
index 66cf1c0e00..1e76473eeb 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java
@@ -99,12 +99,12 @@ public class AccordIncrementalRepairTest extends AccordTestBase
private final List<ExecutedBarrier> barriers = new ArrayList<>();
@Override
- public void onLocalBarrier(@Nonnull Seekables<?, ?> keysOrRanges, @Nonnull Timestamp executeAt)
+ public void onLocalBarrier(@Nonnull Seekables<?, ?> keysOrRanges, @Nonnull TxnId txnId)
{
- super.onLocalBarrier(keysOrRanges, executeAt);
+ super.onLocalBarrier(keysOrRanges, txnId);
synchronized (barriers)
{
- barriers.add(new ExecutedBarrier(keysOrRanges, executeAt));
+ barriers.add(new ExecutedBarrier(keysOrRanges, txnId));
}
}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMigrationTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMigrationTest.java
index a63642f819..97d35d79af 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMigrationTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMigrationTest.java
@@ -26,14 +26,13 @@ import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;
import javax.annotation.Nonnull;
-import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.collect.ImmutableList;
-
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Assert;
@@ -41,12 +40,14 @@ import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.yaml.snakeyaml.Yaml;
+import com.fasterxml.jackson.core.type.TypeReference;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.Config.PaxosVariant;
import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.db.BufferDecoratedKey;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.ConsistencyLevel;
@@ -70,9 +71,10 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.consensus.TransactionalMode;
import org.apache.cassandra.service.consensus.migration.ConsensusKeyMigrationState;
-import org.apache.cassandra.service.consensus.migration.ConsensusRequestRouter;
+import org.apache.cassandra.service.consensus.migration.ConsensusMigratedAt;
import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState;
import org.apache.cassandra.service.consensus.migration.ConsensusMigrationTarget;
+import org.apache.cassandra.service.consensus.migration.ConsensusRequestRouter;
import org.apache.cassandra.service.consensus.migration.TableMigrationState;
import org.apache.cassandra.service.consensus.migration.TransactionalMigrationFromMode;
import org.apache.cassandra.service.paxos.Ballot;
@@ -87,15 +89,13 @@ import org.apache.cassandra.utils.ByteArrayUtil;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JsonUtils;
+import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.PojoToString;
+import org.yaml.snakeyaml.Yaml;
+import static com.google.common.collect.ImmutableList.toImmutableList;
import static java.lang.String.format;
import static java.util.Collections.emptyList;
-import static com.google.common.collect.ImmutableList.toImmutableList;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-
import static org.apache.cassandra.Util.spinUntilSuccess;
import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
import static org.apache.cassandra.db.SystemKeyspace.CONSENSUS_MIGRATION_STATE;
@@ -108,6 +108,9 @@ import static org.apache.cassandra.schema.SchemaConstants.SYSTEM_KEYSPACE_NAME;
import static org.apache.cassandra.service.consensus.migration.ConsensusRequestRouter.ConsensusRoutingDecision.paxosV2;
import static org.apache.cassandra.service.paxos.PaxosState.MaybePromise.Outcome.PROMISE;
import static org.assertj.core.api.Fail.fail;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
/*
* This test suite is intended to serve as an integration test with some pretty good visibility into actual execution
@@ -296,7 +299,7 @@ public class AccordMigrationTest extends AccordTestBase
* Helper to invoke a query and assert that the right metrics change indicating the correct
* paths were taken to execute the query during migration
*/
- private static void assertTargetAccordWrite(Consumer<Integer> query, int coordinatorIndex, int key, int expectedAccordWriteCount, int expectedCasWriteCount, int expectedKeyMigrationCount, int expectedCasBeginRejects, int expectedCasAcceptRejects)
+ private static void assertTargetAccordWrite(Consumer<Integer> query, int coordinatorIndex, int key, List<Pair<ByteBuffer, UUID>> expectedKeyMigrations, int expectedAccordWriteCount, int expectedCasWriteCount, int expectedKeyMigrationCount, int expectedCasBeginRejects, int expectedCasAcceptRejects)
{
int startingWriteCount = getAccordWriteCount(coordinatorIndex);
int startingCasWriteCount = getCasWriteCount(coordinatorIndex);
@@ -304,6 +307,7 @@ public class AccordMigrationTest extends AccordTestBase
int startingCasWriteBeginRejects = getCasWriteBeginRejects(coordinatorIndex);
int startingCasWriteAcceptRejects = getCasWriteAcceptRejects(coordinatorIndex);
query.accept(key);
+ validateKeyMigrations(expectedKeyMigrations);
assertEquals("Accord writes", expectedAccordWriteCount, getAccordWriteCount(coordinatorIndex) - startingWriteCount);
assertEquals("CAS writes", expectedCasWriteCount, getCasWriteCount(coordinatorIndex) - startingCasWriteCount);
assertEquals("Key Migrations", expectedKeyMigrationCount, getKeyMigrationCount(coordinatorIndex) - startingKeyMigrationCount);
@@ -311,7 +315,7 @@ public class AccordMigrationTest extends AccordTestBase
assertEquals("CAS Accept rejects", expectedCasAcceptRejects, getCasWriteAcceptRejects(coordinatorIndex) - startingCasWriteAcceptRejects);
}
- private static Object[][] assertTargetAccordRead(Function<Integer, Object[][]> query, int coordinatorIndex, int key, int expectedAccordReadCount, int expectedCasPrepareCount, int expectedKeyMigrationCount, int expectedCasReadBeginRejects, int expectedCasReadAcceptRejects)
+ private static Object[][] assertTargetAccordRead(Function<Integer, Object[][]> query, int coordinatorIndex, int key, List<Pair<ByteBuffer, UUID>> expectedKeyMigrations, int expectedAccordReadCount, int expectedCasPrepareCount, int expectedKeyMigrationCount, int expectedCasReadBeginRejects, int expectedCasReadAcceptRejects)
{
int startingReadCount = getAccordReadCount(coordinatorIndex);
int startingCasPrepareCount = getCasPrepareCount(coordinatorIndex);
@@ -319,6 +323,7 @@ public class AccordMigrationTest extends AccordTestBase
int startingCasReadBeginRejects = getCasReadBeginRejects(coordinatorIndex);
int startingCasReadAcceptRejects = getCasReadAcceptRejects(coordinatorIndex);
Object[][] result = query.apply(key);
+ validateKeyMigrations(expectedKeyMigrations);
assertEquals("Accord reads", expectedAccordReadCount, getAccordReadCount(coordinatorIndex) - startingReadCount);
assertEquals("CAS prepares", expectedCasPrepareCount, getCasPrepareCount(coordinatorIndex) - startingCasPrepareCount);
assertEquals("Key Migrations", expectedKeyMigrationCount, getKeyMigrationCount(coordinatorIndex) - startingKeyMigrationCount);
@@ -327,7 +332,7 @@ public class AccordMigrationTest extends AccordTestBase
return result;
}
- private static void assertTargetPaxosWrite(Consumer<Integer> query, int coordinatorIndex, int key, int expectedAccordWriteCount, int expectedCasWriteCount, int expectedKeyMigrationCount, int expectedMigrationRejects, int expectedSkippedReads)
+ private static void assertTargetPaxosWrite(Consumer<Integer> query, int coordinatorIndex, int key, List<Pair<ByteBuffer, UUID>> expectedKeyMigrations, int expectedAccordWriteCount, int expectedCasWriteCount, int expectedKeyMigrationCount, int expectedMigrationRejects, int expectedSkippedReads)
{
int startingWriteCount = getAccordWriteCount(coordinatorIndex);
int startingCasWriteCount = getCasWriteCount(coordinatorIndex);
@@ -335,6 +340,7 @@ public class AccordMigrationTest extends AccordTestBase
int startingMigrationRejectsCount = getAccordMigrationRejects(coordinatorIndex);
int startingSkippedReadsCount = getAccordMigrationSkippedReads();
query.accept(key);
+ validateKeyMigrations(expectedKeyMigrations);
assertEquals("Accord writes", expectedAccordWriteCount, getAccordWriteCount(coordinatorIndex) - startingWriteCount);
assertEquals("CAS writes", expectedCasWriteCount, getCasWriteCount(coordinatorIndex) - startingCasWriteCount);
assertEquals("Key Migrations", expectedKeyMigrationCount, getKeyMigrationCount(coordinatorIndex) - startingKeyMigrationCount);
@@ -342,12 +348,60 @@ public class AccordMigrationTest extends AccordTestBase
assertEquals("Accord skipped reads", expectedSkippedReads, getAccordMigrationSkippedReads() - startingSkippedReadsCount);
}
+ private static void validateKeyMigrations(List<Pair<ByteBuffer, UUID>> expectedMigrations)
+ {
+ spinUntilSuccess(() -> {
+ try
+ {
+ List<byte[]> keys = expectedMigrations.stream().map(p -> p.left.array()).collect(Collectors.toList());
+ List<Integer> intKeys = expectedMigrations.stream().map(p -> ByteBufferUtil.toInt(p.left)).collect(Collectors.toList());
+ List<UUID> tables = expectedMigrations.stream().map(p -> p.right).collect(Collectors.toList());
+ for (int i = 1; i < SHARED_CLUSTER.size(); i++)
+ {
+ int instanceIndex = i;
+ IInvokableInstance instance = SHARED_CLUSTER.get(i);
+ instance.runOnInstance(() -> {
+ Map<Pair<ByteBuffer, UUID>, ConsensusMigratedAt> cacheMap = ConsensusKeyMigrationState.MIGRATION_STATE_CACHE.asMap();
+ String cacheMessage = format("Instance %d Expected %s migrations but found in cache %s", instanceIndex, intKeys, cacheMap);
+ assertEquals(cacheMessage, keys.size(), cacheMap.size());
+ for (int j = 0; j < keys.size(); j++)
+ {
+ assertTrue(cacheMessage,
+ cacheMap.containsKey(Pair.create(ByteBuffer.wrap(keys.get(j)), tables.get(j))));
+ }
+
+ UntypedResultSet result = QueryProcessor.executeInternal("SELECT * from " + SYSTEM_KEYSPACE_NAME + "." + CONSENSUS_MIGRATION_STATE);
+ String tableMessage = format("Instance %d Expected %s migrations but found in system table %s", instanceIndex, intKeys, result);
+ assertEquals(tableMessage, keys.size(), result.size());
+ Iterator<UntypedResultSet.Row> resultIterator = result.iterator();
+ for (int j = 0; j < result.size(); j++)
+ {
+ UntypedResultSet.Row row = resultIterator.next();
+ boolean foundKey = false;
+ for (byte[] expectedKey : keys)
+ if (ByteBuffer.wrap(expectedKey).equals(row.getBytes("row_key")))
+ foundKey = true;
+ assertTrue(tableMessage, foundKey);
+ }
+ });
+ }
+ }
+ catch (Throwable t)
+ {
+ // For some reason full stack trace wasn't displayed without rethrowing
+ throw new AssertionError(t);
+ }
+ });
+ }
+
@Test
public void testPaxosToAccordCAS() throws Exception
{
test(format(TABLE_FMT, qualifiedTableName),
cluster -> {
+ List<Pair<ByteBuffer, UUID>> expectedKeyMigrations = new ArrayList<>();
String table = tableName;
+ UUID tableUUID = cluster.get(1).callOnInstance(() -> ColumnFamilyStore.getIfExists(KEYSPACE, table).getTableId().asUUID());
cluster.forEach(node -> node.runOnInstance(() -> {
TableMetadata tbl = Schema.instance.getTableMetadata(KEYSPACE, table);
Assert.assertEquals(TransactionalMode.off, tbl.params.transactionalMode);
@@ -373,24 +427,25 @@ public class AccordMigrationTest extends AccordTestBase
List<Range<Token>> migratingRanges = ImmutableList.of(migratingRange);
// Not actually migrating yet so should do nothing special
- assertTargetAccordWrite(runCasNoApply, 1, migratingKey, 0, 1, 0, 0, 0);
+ assertTargetAccordWrite(runCasNoApply, 1, migratingKey, expectedKeyMigrations, 0, 1, 0, 0, 0);
// Mark ranges migrating and check migration state is correct
nodetool(coordinator, "consensus_admin", "begin-migration", "-st", midToken.toString(), "-et", maxToken.toString(), "-tp", "accord", KEYSPACE, tableName);
assertMigrationState(tableName, ConsensusMigrationTarget.accord, emptyList(), migratingRanges, 1);
// Should be routed directly to Accord, and perform key migration, as well as key migration read in Accord
- assertTargetAccordWrite(runCasNoApply, 1, migratingKey, 1, 0, 1, 0, 0);
+ addExpectedMigratedKey(expectedKeyMigrations, migratingKey, tableUUID);
+ assertTargetAccordWrite(runCasNoApply, 1, migratingKey, expectedKeyMigrations, 1, 0, 1, 0, 0);
// Should not repeat key migration, and should still do a migration read in Accord
- assertTargetAccordWrite(runCasNoApply, 1, migratingKey, 1, 0, 0, 0, 0);
+ assertTargetAccordWrite(runCasNoApply, 1, migratingKey, expectedKeyMigrations, 1, 0, 0, 0, 0);
// Should run on Paxos since it is not in the migrating range
- assertTargetAccordWrite(runCasNoApply, 1, notMigratingKey, 0, 1, 0, 0, 0);
+ assertTargetAccordWrite(runCasNoApply, 1, notMigratingKey, expectedKeyMigrations, 0, 1, 0, 0, 0);
// Check that the coordinator on the other node also has saved that the key migration was performed
// and runs the query on Accord immediately without key migration
- assertTargetAccordWrite(runCasOnSecondNode, 2, migratingKey, 1, 0, 0, 0, 0);
+ assertTargetAccordWrite(runCasOnSecondNode, 2, migratingKey, expectedKeyMigrations, 1, 0, 0, 0, 0);
// Forced repair while a node is down shouldn't work, use repair instead of finish-migration because repair exposes --force
// and regular Cassandra repairs are eligible to drive migration so it's important they check --force and down nodes
@@ -411,10 +466,10 @@ public class AccordMigrationTest extends AccordTestBase
assertMigrationState(tableName, ConsensusMigrationTarget.accord, migratingRanges, emptyList(), 0);
// Should run on Accord, and not perform key migration nor should it need to perform a migration read in Accord now that it is repaired
- assertTargetAccordWrite(runCasNoApply, 1, migratingKey, 1, 0, 0, 0, 0);
+ assertTargetAccordWrite(runCasNoApply, 1, migratingKey, expectedKeyMigrations, 1, 0, 0, 0, 0);
// Should run on Paxos, and not perform key migration
- assertTargetAccordWrite(runCasNoApply, 1, notMigratingKey, 0, 1, 0, 0, 0);
+ assertTargetAccordWrite(runCasNoApply, 1, notMigratingKey, expectedKeyMigrations, 0, 1, 0, 0, 0);
// Pivot to testing repair with a subrange of the migrating range as well as key migration
// Will use the unmigrated range between lowerMidToken and midToken
@@ -429,7 +484,8 @@ public class AccordMigrationTest extends AccordTestBase
saveAcceptedPaxosProposal(tableName, ballotString, migratingKey);
// PaxosRepair will have inserted a condition matching row, so it can apply, demonstrating repair and
// key migration occurred
- assertTargetAccordWrite(runCasApplies, 1, migratingKey, 1, 0, 1, 0, 0);
+ addExpectedMigratedKey(expectedKeyMigrations, migratingKey, tableUUID);
+ assertTargetAccordWrite(runCasApplies, 1, migratingKey, expectedKeyMigrations, 1, 0, 1, 0, 0);
// This will force the write to use the normal write patch
cluster.get(1).runOnInstance(() -> ConsensusRequestRouter.setInstance(new PaxosToAccordMigrationNotHappeningUpToBegin()));
@@ -443,7 +499,8 @@ public class AccordMigrationTest extends AccordTestBase
// This will force the request to run on Paxos up to Accept
// and the accept will be rejected at both nodes and we are certain we need to retry the transaction
cluster.get(1).runOnInstance(() -> ConsensusRequestRouter.setInstance(new PaxosToAccordMigrationNotHappeningUpToBegin()));
- assertTargetAccordWrite(runCasApplies, 1, migratingKey, 1, 1, 1, 0, 1);
+ addExpectedMigratedKey(expectedKeyMigrations, migratingKey, tableUUID);
+ assertTargetAccordWrite(runCasApplies, 1, migratingKey, expectedKeyMigrations, 1, 1, 1, 0, 1);
// One node will now accept the other will reject and we are uncertain if we should retry the transaction
// and should surface that as a timeout exception
@@ -467,7 +524,9 @@ public class AccordMigrationTest extends AccordTestBase
// retry it on Accord
cluster.get(1).runOnInstance(() -> ConsensusRequestRouter.setInstance(new RoutesToPaxosOnce()));
// Should exit Paxos from begin, key migration should occur because it's a new key, and Accord will need to do a migration read
- assertTargetAccordWrite(runCasNoApply, 1, testingKeys.next(), 1, 1, 1, 1, 0);
+ migratingKey = testingKeys.next();
+ addExpectedMigratedKey(expectedKeyMigrations, migratingKey, tableUUID);
+ assertTargetAccordWrite(runCasNoApply, 1, migratingKey, expectedKeyMigrations, 1, 1, 1, 1, 0);
// Now do two repairs to complete the migration repair, and we are done with black box integration testing
// First repair is a range smack dab in the middle
@@ -494,6 +553,9 @@ public class AccordMigrationTest extends AccordTestBase
{
test(format(TABLE_FMT, qualifiedTableName),
cluster -> {
+ String table = tableName;
+ UUID tableUUID = cluster.get(1).callOnInstance(() -> ColumnFamilyStore.getIfExists(KEYSPACE, table).getTableId().asUUID());
+ List<Pair<ByteBuffer, UUID>> expectedKeyMigrations = new ArrayList<>();
cluster.schemaChange(format("ALTER TABLE %s.%s WITH transactional_mode='%s'", KEYSPACE, tableName, TransactionalMode.full));
String readCQL = format("SELECT * FROM %s WHERE id = ? and c = %s", qualifiedTableName, CLUSTERING_VALUE);
Function<Integer, Object[][]> runRead = key -> cluster.coordinator(1).execute(readCQL, SERIAL, key);
@@ -501,17 +563,21 @@ public class AccordMigrationTest extends AccordTestBase
List<Range<Token>> migratingRanges = ImmutableList.of(migratingRange);
int key = 0;
- assertTargetAccordRead(runRead, 1, 0, 0, 1, 0, 0, 0);
+ assertTargetAccordRead(runRead, 1, key, expectedKeyMigrations, 0, 1, 0, 0, 0);
// Mark wrap around range as migrating
nodetool(coordinator, "consensus_admin", "begin-migration", "-st", String.valueOf(Long.MIN_VALUE + 1), "-et", String.valueOf(Long.MIN_VALUE), "-tp", "accord", KEYSPACE, tableName);
assertMigrationState(tableName, ConsensusMigrationTarget.accord, emptyList(), migratingRanges, 1);
// Should run directly on accord, migrate the key, and perform a quorum read from Accord, Paxos repair will run prepare once
- assertTargetAccordRead(runRead, 1, key++, 1, 1, 1, 0, 0);
+ addExpectedMigratedKey(expectedKeyMigrations, key, tableUUID);
+ assertTargetAccordRead(runRead, 1, key, expectedKeyMigrations, 1, 1, 1, 0, 0);
+ key++;
// Should run up to accept with both nodes refusing to accept
savePromisedAndCommittedPaxosProposal(tableName, key);
cluster.get(1).runOnInstance(() -> ConsensusRequestRouter.setInstance(new PaxosToAccordMigrationNotHappeningUpToBegin()));
- assertTargetAccordRead(runRead, 1, key++, 1, 2, 1, 0, 1);
+ addExpectedMigratedKey(expectedKeyMigrations, key, tableUUID);
+ assertTargetAccordRead(runRead, 1, key, expectedKeyMigrations, 1, 2, 1, 0, 1);
+ key++;
});
}
@@ -542,6 +608,7 @@ public class AccordMigrationTest extends AccordTestBase
String casCQL = format(CAS_FMT, qualifiedTableName, CLUSTERING_VALUE);
Consumer<Integer> runCasNoApply = key -> assertRowEquals(cluster, new Object[]{false}, casCQL, key);
String tableName = qualifiedTableName.split("\\.")[1];
+ UUID tableUUID = cluster.get(1).callOnInstance(() -> ColumnFamilyStore.getIfExists(KEYSPACE, tableName).getTableId().asUUID());
alterTableTransactionalMode(TransactionalMode.mixed_reads);
assertTransactionalModes(TransactionalMode.mixed_reads, TransactionalMigrationFromMode.off);
@@ -562,25 +629,33 @@ public class AccordMigrationTest extends AccordTestBase
Iterator<Integer> paxosMigratingKeys = getKeysBetweenTokens(upperMidToken, maxToken);
Iterator<Integer> accordKeys = getKeysBetweenTokens(midToken, upperMidToken);
+ List<Pair<ByteBuffer, UUID>> expectedKeyMigrations = new ArrayList<>();
+
// Paxos non-migrating keys should run on Paxos as per normal
- assertTargetPaxosWrite(runCasNoApply, 1, paxosNonMigratingKeys.next(), 0, 1, 0, 0, 0);
+ assertTargetPaxosWrite(runCasNoApply, 1, paxosNonMigratingKeys.next(), expectedKeyMigrations, 0, 1, 0, 0, 0);
+ Integer nextMigratingKey = paxosMigratingKeys.next();
+ addExpectedMigratedKey(expectedKeyMigrations, nextMigratingKey, tableUUID);
// Paxos migrating keys should be key migrated which means a local barrier is run by Paxos during read at each replica, the key migration barrier is also counted as a write
- assertTargetPaxosWrite(runCasNoApply, 1, paxosMigratingKeys.next(), 1, 1, 1, 0, 0);
+ assertTargetPaxosWrite(runCasNoApply, 1, nextMigratingKey, expectedKeyMigrations, 1, 1, 1, 0, 0);
// A key from a range migrated to Accord is now not migrating/migrated and should be accessed through Accord
- assertTargetPaxosWrite(runCasNoApply, 1, accordKeys.next(), 1, 0, 0, 0, 0);
+ assertTargetPaxosWrite(runCasNoApply, 1, accordKeys.next(), expectedKeyMigrations, 1, 0, 0, 0, 0);
// If an Accord transaction races with cluster metadata updates it should be rejected if the epoch it runs in contains the migration
cluster.get(1).runOnInstance(() -> ConsensusRequestRouter.setInstance(new RoutesToAccordOnce()));
- assertTargetPaxosWrite(runCasNoApply, 1, paxosMigratingKeys.next(), 2, 1, 1, 1, 1);
+ nextMigratingKey = paxosMigratingKeys.next();
+ addExpectedMigratedKey(expectedKeyMigrations, nextMigratingKey, tableUUID);
+ assertTargetPaxosWrite(runCasNoApply, 1, nextMigratingKey, expectedKeyMigrations, 2, 1, 1, 1, 1);
// Repair the currently migrating range from when targets were switched, but it's not an Accord repair, this is to make sure the wrong repair type doesn't trigger progress
nodetool(coordinator, "repair", "-st", upperMidToken.toString(), "-et", maxAlignedWithLocalRanges.toString(), "--paxos-only");
assertMigrationState(tableName, ConsensusMigrationTarget.paxos, ImmutableList.of(new Range(minToken, midToken), new Range(maxToken, minToken)), ImmutableList.of(accordMigratingRange), 1);
// Paxos migrating keys should still need key migration after non-Accord repair
- assertTargetPaxosWrite(runCasNoApply, 1, paxosMigratingKeys.next(), 1, 1, 1, 0, 0);
+ nextMigratingKey = paxosMigratingKeys.next();
+ addExpectedMigratedKey(expectedKeyMigrations, nextMigratingKey, tableUUID);
+ assertTargetPaxosWrite(runCasNoApply, 1, nextMigratingKey, expectedKeyMigrations, 1, 1, 1, 0, 0);
// Now do it with an Accord repair so key migration shouldn't be necessary
nodetool(coordinator, "consensus_admin", "finish-migration", "-st", upperMidToken.toString(), "-et", maxAlignedWithLocalRanges.toString());
@@ -591,40 +666,16 @@ public class AccordMigrationTest extends AccordTestBase
assertMigrationState(tableName, ConsensusMigrationTarget.paxos, ImmutableList.of(new Range(minToken, midToken), repairedRange, new Range(maxToken, minToken)), ImmutableList.of(remainingRange), 1);
// Paxos migrating keys shouldn't need key migration after Accord repair
- assertTargetPaxosWrite(runCasNoApply, 1, paxosMigratingKeys.next(), 0, 1, 0, 0, 0);
+ assertTargetPaxosWrite(runCasNoApply, 1, paxosMigratingKeys.next(), expectedKeyMigrations, 0, 1, 0, 0, 0);
});
}
- private static void assertCompletedMigrationState(String tableName) throws Throwable
+ private static void addExpectedMigratedKey(List<Pair<ByteBuffer, UUID>> expectedKeyMigrations, Integer nextMigratingKey, UUID tableUUID)
{
- // Validate nodetool consensus admin list output
- String yamlResultString = nodetool(SHARED_CLUSTER.coordinator(1), "consensus_admin", "list");
- Map<String, Object> yamlStateMap = new Yaml().load(yamlResultString);
- String minifiedYamlResultString = nodetool(SHARED_CLUSTER.coordinator(1), "consensus_admin", "list", "-f", "minified-yaml");
- Map<String, Object> minifiedYamlStateMap = new Yaml().load(minifiedYamlResultString);
- String jsonResultString = nodetool(SHARED_CLUSTER.coordinator(1), "consensus_admin", "list", "-f", "json");
- Map<String, Object> jsonStateMap = JsonUtils.JSON_OBJECT_MAPPER.readValue(jsonResultString, new TypeReference<Map<String, Object>>(){});
- String minifiedJsonResultString = nodetool(SHARED_CLUSTER.coordinator(1), "consensus_admin", "list", "-f", "minified-json");
- Map<String, Object> minifiedJsonStateMap = JsonUtils.JSON_OBJECT_MAPPER.readValue(minifiedJsonResultString, new TypeReference<Map<String, Object>>(){});
-
- for (Map<String, Object> migrationStateMap : ImmutableList.of(yamlStateMap, jsonStateMap, minifiedYamlStateMap, minifiedJsonStateMap)) {
- assertEquals(PojoToString.CURRENT_VERSION, migrationStateMap.get("version"));
- assertTrue(Epoch.EMPTY.getEpoch() < ((Number) migrationStateMap.get("lastModifiedEpoch")).longValue());
- List<Map<String, Object>> tableStates = (List<Map<String, Object>>) migrationStateMap.get("tableStates");
- assertEquals(0, tableStates.size());
- }
- spinUntilSuccess(() -> {
- for (IInvokableInstance instance : SHARED_CLUSTER)
- {
- ConsensusMigrationState snapshot = getMigrationStateSnapshot(instance);
- assertEquals(0, snapshot.tableStates.size());
- instance.runOnInstance(() -> {
- TableMetadata tbl = Schema.instance.getTableMetadata(KEYSPACE, tableName);
- Assert.assertEquals(TransactionalMigrationFromMode.none, tbl.params.transactionalMigrationFrom);
- });
- }
- });
+ ByteBuffer key = ByteBuffer.allocate(4);
+ key.putInt(0, nextMigratingKey);
+ expectedKeyMigrations.add(Pair.create(key, tableUUID));
}
private static void assertMigrationState(String tableName, ConsensusMigrationTarget target, List<Range<Token>> migratedRanges, List<Range<Token>> migratingRanges, int numMigratingEpochs) throws Throwable
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org