You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by dc...@apache.org on 2024/09/28 06:41:46 UTC
(cassandra) branch cep-15-accord updated: (Accord): C* stores table in Range which will cause ranges to be removed from Accord when DROP TABLE is performed
This is an automated email from the ASF dual-hosted git repository.
dcapwell pushed a commit to branch cep-15-accord
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cep-15-accord by this push:
new c4d0f8cb5d (Accord): C* stores table in Range which will cause ranges to be removed from Accord when DROP TABLE is performed
c4d0f8cb5d is described below
commit c4d0f8cb5ddd9e9ac98a88dd159e314e5e76a97d
Author: David Capwell <dc...@apache.org>
AuthorDate: Wed Sep 4 15:59:35 2024 -0700
(Accord): C* stores table in Range which will cause ranges to be removed from Accord when DROP TABLE is performed
patch by David Capwell, Sam Tunnicliffe; reviewed by Sam Tunnicliffe for CASSANDRA-18675
---
modules/accord | 2 +-
.../cql3/statements/TransactionStatement.java | 6 +
.../statements/schema/AlterSchemaStatement.java | 7 +-
.../statements/schema/AlterTableStatement.java | 3 +
.../statements/schema/CreateViewStatement.java | 5 +
.../statements/schema/DropKeyspaceStatement.java | 29 +-
.../cql3/statements/schema/DropTableStatement.java | 27 ++
.../apache/cassandra/schema/KeyspaceParams.java | 8 +-
.../org/apache/cassandra/schema/TableParams.java | 61 ++--
.../service/accord/AccordSafeCommandStore.java | 9 +
.../cassandra/service/accord/AccordService.java | 252 +++++++++++++++-
.../service/accord/AccordStaleReplicas.java | 2 +-
.../cassandra/service/accord/AccordTopology.java | 7 +-
.../cassandra/service/accord/IAccordService.java | 5 +
.../fastpath/InheritKeyspaceFastPathStrategy.java | 2 +-
.../fastpath/ParameterizedFastPathStrategy.java | 8 +-
.../accord/fastpath/SimpleFastPathStrategy.java | 2 +-
.../migration/ConsensusMigrationState.java | 3 +
.../org/apache/cassandra/tcm/CMSOperations.java | 17 ++
.../apache/cassandra/tcm/CMSOperationsMBean.java | 1 +
.../org/apache/cassandra/tcm/ClusterMetadata.java | 26 +-
.../org/apache/cassandra/tcm/MetadataKeys.java | 39 ++-
.../apache/cassandra/tcm/MultiStepOperation.java | 4 +-
.../cassandra/tcm/StubClusterMetadataService.java | 11 +-
.../org/apache/cassandra/tcm/Transformation.java | 5 +-
.../cassandra/tcm/membership/NodeVersion.java | 2 +-
.../cassandra/tcm/sequences/DropAccordTable.java | 324 +++++++++++++++++++++
.../tcm/sequences/InProgressSequences.java | 12 +-
.../cassandra/tcm/serialization/Version.java | 27 +-
.../cassandra/tcm/transformations/AlterSchema.java | 2 +-
.../tcm/transformations/FinishDropAccordTable.java | 143 +++++++++
.../transformations/PrepareDropAccordTable.java | 115 ++++++++
src/java/org/apache/cassandra/tools/NodeTool.java | 3 +-
.../apache/cassandra/tools/nodetool/CMSAdmin.java | 12 +
.../cassandra/distributed/shared/ClusterUtils.java | 10 +
.../test/accord/AccordDropKeyspaceTest.java | 63 ++++
.../test/accord/AccordDropTableBase.java | 159 ++++++++++
.../test/accord/AccordDropTableTest.java | 63 ++++
.../test/log/ClusterMetadataTestHelper.java | 32 +-
.../fuzz/topology/AccordTopologyMixupTest.java | 21 ++
.../fuzz/topology/TopologyMixupTestBase.java | 18 ++
.../cassandra/cql3/PreparedStatementsTest.java | 6 +
.../apache/cassandra/schema/TableParamsTest.java | 52 ++++
.../service/accord/AccordKeyspaceTest.java | 2 +-
.../service/accord/AccordStaleReplicasTest.java | 2 +-
.../service/accord/AccordSyncPropagatorTest.java | 9 +-
.../cassandra/service/accord/EpochSyncTest.java | 75 +++--
.../service/accord/FetchMinEpochTest.java | 1 +
.../AccordRoutingKeyByteSourceTest.java | 2 +-
.../accord/serializers/DepsSerializerTest.java | 20 +-
.../tcm/ClusterMetadataMetadataKeyTest.java | 100 +++++++
.../tcm/ClusterMetadataSerializerTest.java | 83 ++++++
.../tcm/ClusterMetadataTransformationTest.java | 4 +-
.../tcm/ValidatingClusterMetadataService.java | 107 +++++++
.../tcm/log/LogListenerNotificationTest.java | 2 +-
.../tcm/sequences/DropAccordTableTest.java | 223 ++++++++++++++
.../transformations/AccordMarkRejoiningTest.java | 6 +-
.../tcm/transformations/AccordMarkStaleTest.java | 6 +-
.../cassandra/utils/CassandraGenerators.java | 279 ++++++++++++++++--
.../cassandra/utils/CassandraGeneratorsTest.java | 2 +-
.../org/apache/cassandra/utils/Generators.java | 25 ++
.../cassandra/utils/SimulatedMiniCluster.java | 8 +-
62 files changed, 2350 insertions(+), 211 deletions(-)
diff --git a/modules/accord b/modules/accord
index 107966eeaf..4844e64945 160000
--- a/modules/accord
+++ b/modules/accord
@@ -1 +1 @@
-Subproject commit 107966eeaf69a968dcd736519b6d6199ee28acb1
+Subproject commit 4844e64945b720c802dce11d811e25665f9da826
diff --git a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java
index f1dcdb4eef..6d59a3a37d 100644
--- a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java
@@ -101,6 +101,7 @@ public class TransactionStatement implements CQLStatement.CompositeCQLStatement,
public static final String NO_CONDITIONS_IN_UPDATES_MESSAGE = "Updates within transactions may not specify their own conditions; %s statement %s";
public static final String NO_TIMESTAMPS_IN_UPDATES_MESSAGE = "Updates within transactions may not specify custom timestamps; %s statement %s";
public static final String TRANSACTIONS_DISABLED_ON_TABLE_MESSAGE = "Accord transactions are disabled on table (See transactional_mode in table options); %s statement %s";
+ public static final String TRANSACTIONS_DISABLED_ON_TABLE_BEING_DROPPED_MESSAGE = "Accord transactions are disabled on table (table is being dropped); %s statement %s";
public static final String NO_COUNTERS_IN_TXNS_MESSAGE = "Counter columns cannot be accessed within a transaction; %s statement %s";
public static final String EMPTY_TRANSACTION_MESSAGE = "Transaction contains no reads or writes";
public static final String SELECT_REFS_NEED_COLUMN_MESSAGE = "SELECT references must specify a column.";
@@ -531,6 +532,8 @@ public class TransactionStatement implements CQLStatement.CompositeCQLStatement,
if (!prepared.table.isAccordEnabled())
throw invalidRequest(TRANSACTIONS_DISABLED_ON_TABLE_MESSAGE, "SELECT", prepared.source);
+ if (prepared.table.params.pendingDrop)
+ throw invalidRequest(TRANSACTIONS_DISABLED_ON_TABLE_BEING_DROPPED_MESSAGE, "SELECT", prepared.source);
if (prepared.table.isCounter())
throw invalidRequest(NO_COUNTERS_IN_TXNS_MESSAGE, "SELECT", prepared.source);
@@ -551,6 +554,8 @@ public class TransactionStatement implements CQLStatement.CompositeCQLStatement,
if (!prepared.table.isAccordEnabled())
throw invalidRequest(TRANSACTIONS_DISABLED_ON_TABLE_MESSAGE, "SELECT", prepared.source);
+ if (prepared.table.params.pendingDrop)
+ throw invalidRequest(TRANSACTIONS_DISABLED_ON_TABLE_BEING_DROPPED_MESSAGE, "SELECT", prepared.source);
if (prepared.table.isCounter())
throw invalidRequest(NO_COUNTERS_IN_TXNS_MESSAGE, "SELECT", prepared.source);
@@ -577,6 +582,7 @@ public class TransactionStatement implements CQLStatement.CompositeCQLStatement,
ModificationStatement prepared = parsed.prepare(state, bindVariables);
checkTrue(prepared.metadata().isAccordEnabled(), TRANSACTIONS_DISABLED_ON_TABLE_MESSAGE, prepared.type, prepared.source);
+ checkFalse(prepared.metadata().params.pendingDrop, TRANSACTIONS_DISABLED_ON_TABLE_BEING_DROPPED_MESSAGE, prepared.type, prepared.source);
checkFalse(prepared.hasConditions(), NO_CONDITIONS_IN_UPDATES_MESSAGE, prepared.type, prepared.source);
checkFalse(prepared.isTimestampSet(), NO_TIMESTAMPS_IN_UPDATES_MESSAGE, prepared.type, prepared.source);
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/AlterSchemaStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/AlterSchemaStatement.java
index 79b4298550..cfab82f73f 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/AlterSchemaStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/AlterSchemaStatement.java
@@ -160,7 +160,7 @@ abstract public class AlterSchemaStatement implements CQLStatement.SingleKeyspac
ClusterMetadata metadata = ClusterMetadata.current();
apply(metadata);
- ClusterMetadata result = Schema.instance.submit(this);
+ ClusterMetadata result = commit(metadata);
KeyspacesDiff diff = Keyspaces.diff(metadata.schema.getKeyspaces(), result.schema.getKeyspaces());
clientWarnings(diff).forEach(ClientWarn.instance::warn);
@@ -185,6 +185,11 @@ abstract public class AlterSchemaStatement implements CQLStatement.SingleKeyspac
return new ResultMessage.SchemaChange(schemaChangeEvent(diff));
}
+ protected ClusterMetadata commit(ClusterMetadata metadata)
+ {
+ return Schema.instance.submit(this);
+ }
+
private void validateKeyspaceName()
{
if (!SchemaConstants.isValidName(keyspaceName))
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java
index fa00b7d954..b3536910c8 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java
@@ -123,6 +123,9 @@ public abstract class AlterTableStatement extends AlterSchemaStatement
return schema;
}
+ if (table.params.pendingDrop)
+ throw ire("Cannot use ALTER TABLE on a table that is being dropped.");
+
if (table.isView())
throw ire("Cannot use ALTER TABLE on a materialized view; use ALTER MATERIALIZED VIEW instead");
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateViewStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateViewStatement.java
index f6c8135bb1..e4ebd36917 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateViewStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/CreateViewStatement.java
@@ -174,6 +174,11 @@ public final class CreateViewStatement extends AlterSchemaStatement
viewName, tableName);
}
+ if (table.params.pendingDrop)
+ throw ire("Cannot create materialized view '%s' for base table " +
+ "'%s' as it is being dropped.",
+ viewName, tableName);
+
/*
* Process SELECT clause
*/
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java
index e074da54a3..e7159cda8d 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java
@@ -17,13 +17,18 @@
*/
package org.apache.cassandra.cql3.statements.schema;
+import java.util.List;
+import java.util.stream.Collectors;
+
import org.apache.cassandra.audit.AuditLogContext;
import org.apache.cassandra.audit.AuditLogEntryType;
import org.apache.cassandra.auth.Permission;
import org.apache.cassandra.cql3.CQLStatement;
import org.apache.cassandra.db.guardrails.Guardrails;
+import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.Keyspaces;
import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff;
+import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.transport.Event.SchemaChange;
@@ -45,8 +50,30 @@ public final class DropKeyspaceStatement extends AlterSchemaStatement
Guardrails.dropKeyspaceEnabled.ensureEnabled(state);
Keyspaces schema = metadata.schema.getKeyspaces();
- if (schema.containsKeyspace(keyspaceName))
+ KeyspaceMetadata keyspace = schema.getNullable(keyspaceName);
+ if (keyspace != null)
+ {
+ // check that no accord tables in the keyspace are currently in the process of being dropped
+ List<TableMetadata> pendingDrop = keyspace.tables.stream()
+ .filter(t -> t.params.pendingDrop)
+ .collect(Collectors.toList());
+ if (!pendingDrop.isEmpty())
+ throw ire("Cannot drop keyspace '%s' as it contains accord tables which are currently being dropped. " +
+ "Please wait for those operations to complete before dropping the keyspace. (%s)",
+ keyspaceName, pendingDrop.stream()
+ .map(Object::toString)
+ .collect(Collectors.joining(",")));
+
+ List<TableMetadata> accordTables = keyspace.tables.stream()
+ .filter(TableMetadata::isAccordEnabled)
+ .collect(Collectors.toList());
+ if (!accordTables.isEmpty())
+ throw ire("Cannot drop keyspace '%s' as it contains accord tables. (%s)",
+ keyspaceName, accordTables.stream()
+ .map(Object::toString)
+ .collect(Collectors.joining(",")));
return schema.without(keyspaceName);
+ }
if (ifExists)
return schema;
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java
index 56848a8c22..4a5371611a 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java
@@ -27,6 +27,10 @@ import org.apache.cassandra.schema.*;
import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.ClusterMetadataService;
+import org.apache.cassandra.tcm.sequences.InProgressSequences;
+import org.apache.cassandra.tcm.transformations.PrepareDropAccordTable;
+import org.apache.cassandra.tcm.sequences.DropAccordTable.TableReference;
import org.apache.cassandra.transport.Event.SchemaChange;
import org.apache.cassandra.transport.Event.SchemaChange.Change;
import org.apache.cassandra.transport.Event.SchemaChange.Target;
@@ -48,6 +52,26 @@ public final class DropTableStatement extends AlterSchemaStatement
this.ifExists = ifExists;
}
+ @Override
+ protected ClusterMetadata commit(ClusterMetadata metadata)
+ {
+ KeyspaceMetadata keyspace = metadata.schema.getKeyspaces().getNullable(keyspaceName);
+ TableMetadata table = null == keyspace
+ ? null
+ : keyspace.getTableOrViewNullable(tableName);
+ if (table == null // this can happen when ifExists=true... since its already been validated can skip
+ || !table.isAccordEnabled())
+ return super.commit(metadata);
+
+ // Multi-Step Operation
+ // 1) mark the table as pending delete
+ // 2) await for Accord to finish transactions
+ // 3) drop table
+ TableReference ref = TableReference.from(table);
+ ClusterMetadataService.instance().commit(new PrepareDropAccordTable(ref));
+ return InProgressSequences.finishInProgressSequences(ref);
+ }
+
public Keyspaces apply(ClusterMetadata metadata)
{
Guardrails.dropTruncateTableEnabled.ensureEnabled(state);
@@ -70,6 +94,9 @@ public final class DropTableStatement extends AlterSchemaStatement
if (table.isView())
throw ire("Cannot use DROP TABLE on a materialized view. Please use DROP MATERIALIZED VIEW instead.");
+ if (table.isAccordEnabled() && table.params.pendingDrop)
+ throw ire("Table '%s.%s' is already being dropped", keyspaceName, tableName);
+
Iterable<ViewMetadata> views = keyspace.views.forTable(table.id);
if (!isEmpty(views))
{
diff --git a/src/java/org/apache/cassandra/schema/KeyspaceParams.java b/src/java/org/apache/cassandra/schema/KeyspaceParams.java
index a2ac768685..99d87b3c7b 100644
--- a/src/java/org/apache/cassandra/schema/KeyspaceParams.java
+++ b/src/java/org/apache/cassandra/schema/KeyspaceParams.java
@@ -33,7 +33,7 @@ import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.serialization.MetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
-import static org.apache.cassandra.tcm.serialization.Version.V2;
+import static org.apache.cassandra.tcm.serialization.Version.MIN_ACCORD_VERSION;
/**
* An immutable class representing keyspace parameters (durability and replication).
@@ -157,7 +157,7 @@ public final class KeyspaceParams
{
ReplicationParams.serializer.serialize(t.replication, out, version);
out.writeBoolean(t.durableWrites);
- if (version.isAtLeast(V2))
+ if (version.isAtLeast(MIN_ACCORD_VERSION))
FastPathStrategy.serializer.serialize(t.fastPath, out, version);
}
@@ -165,7 +165,7 @@ public final class KeyspaceParams
{
ReplicationParams params = ReplicationParams.serializer.deserialize(in, version);
boolean durableWrites = in.readBoolean();
- FastPathStrategy fastPath = version.isAtLeast(V2)
+ FastPathStrategy fastPath = version.isAtLeast(MIN_ACCORD_VERSION)
? FastPathStrategy.serializer.deserialize(in, version)
: FastPathStrategy.simple();
return new KeyspaceParams(durableWrites, params, fastPath);
@@ -175,7 +175,7 @@ public final class KeyspaceParams
{
return ReplicationParams.serializer.serializedSize(t.replication, version) +
TypeSizes.sizeof(t.durableWrites) +
- (version.isAtLeast(V2) ? FastPathStrategy.serializer.serializedSize(t.fastPath, version) : 0);
+ (version.isAtLeast(MIN_ACCORD_VERSION) ? FastPathStrategy.serializer.serializedSize(t.fastPath, version) : 0);
}
}
}
diff --git a/src/java/org/apache/cassandra/schema/TableParams.java b/src/java/org/apache/cassandra/schema/TableParams.java
index c836da9b39..7eff48cc76 100644
--- a/src/java/org/apache/cassandra/schema/TableParams.java
+++ b/src/java/org/apache/cassandra/schema/TableParams.java
@@ -74,7 +74,8 @@ public final class TableParams
READ_REPAIR,
FAST_PATH,
TRANSACTIONAL_MODE,
- TRANSACTIONAL_MIGRATION_FROM;
+ TRANSACTIONAL_MIGRATION_FROM,
+ PENDING_DROP;
@Override
public String toString()
@@ -105,6 +106,7 @@ public final class TableParams
public final FastPathStrategy fastPath;
public final TransactionalMode transactionalMode;
public final TransactionalMigrationFromMode transactionalMigrationFrom;
+ public final boolean pendingDrop;
private TableParams(Builder builder)
{
@@ -132,6 +134,7 @@ public final class TableParams
fastPath = builder.fastPath;
transactionalMode = builder.transactionalMode != null ? builder.transactionalMode : TransactionalMode.off;
transactionalMigrationFrom = builder.transactionalMigrationFrom;
+ pendingDrop = builder.pendingDrop;
}
public static Builder builder()
@@ -162,7 +165,8 @@ public final class TableParams
.readRepair(params.readRepair)
.fastPath(params.fastPath)
.transactionalMode(params.transactionalMode)
- .transactionalMigrationFrom(params.transactionalMigrationFrom);
+ .transactionalMigrationFrom(params.transactionalMigrationFrom)
+ .pendingDrop(params.pendingDrop);
}
public Builder unbuild()
@@ -256,7 +260,8 @@ public final class TableParams
&& readRepair == p.readRepair
&& fastPath.equals(fastPath)
&& transactionalMode == p.transactionalMode
- && transactionalMigrationFrom == p.transactionalMigrationFrom;
+ && transactionalMigrationFrom == p.transactionalMigrationFrom
+ && pendingDrop == p.pendingDrop;
}
@Override
@@ -283,7 +288,8 @@ public final class TableParams
readRepair,
fastPath,
transactionalMode,
- transactionalMigrationFrom);
+ transactionalMigrationFrom,
+ pendingDrop);
}
@Override
@@ -313,6 +319,7 @@ public final class TableParams
.add(Option.FAST_PATH.toString(), fastPath)
.add(Option.TRANSACTIONAL_MODE.toString(), transactionalMode)
.add(Option.TRANSACTIONAL_MIGRATION_FROM.toString(), transactionalMigrationFrom)
+ .add(PENDING_DROP.toString(), pendingDrop)
.toString();
}
@@ -400,6 +407,7 @@ public final class TableParams
private FastPathStrategy fastPath = FastPathStrategy.inheritKeyspace();
private TransactionalMode transactionalMode = TransactionalMode.off;
public TransactionalMigrationFromMode transactionalMigrationFrom = TransactionalMigrationFromMode.none;
+ public boolean pendingDrop = false;
public Builder()
{
@@ -541,6 +549,12 @@ public final class TableParams
extensions = ImmutableMap.copyOf(val);
return this;
}
+
+ public Builder pendingDrop(boolean pendingDrop)
+ {
+ this.pendingDrop = pendingDrop;
+ return this;
+ }
}
public static class Serializer implements MetadataSerializer<TableParams>
@@ -558,18 +572,20 @@ public final class TableParams
out.writeUTF(t.speculativeRetry.toString());
out.writeUTF(t.additionalWritePolicy.toString());
if (version.isAtLeast(Version.V2))
- {
out.writeUTF(t.memtable.configurationKey());
- FastPathStrategy.serializer.serialize(t.fastPath, out, version);
- }
serializeMap(t.caching.asMap(), out);
serializeMap(t.compaction.asMap(), out);
serializeMap(t.compression.asMap(), out);
serializeMapBB(t.extensions, out);
out.writeBoolean(t.cdc);
out.writeUTF(t.readRepair.name());
- out.writeInt(t.transactionalMode.ordinal());
- out.writeInt(t.transactionalMigrationFrom.ordinal());
+ if (version.isAtLeast(Version.MIN_ACCORD_VERSION))
+ {
+ FastPathStrategy.serializer.serialize(t.fastPath, out, version);
+ out.writeInt(t.transactionalMode.ordinal());
+ out.writeInt(t.transactionalMigrationFrom.ordinal());
+ out.writeBoolean(t.pendingDrop);
+ }
}
public TableParams deserialize(DataInputPlus in, Version version) throws IOException
@@ -586,21 +602,25 @@ public final class TableParams
.speculativeRetry(SpeculativeRetryPolicy.fromString(in.readUTF()))
.additionalWritePolicy(SpeculativeRetryPolicy.fromString(in.readUTF()))
.memtable(version.isAtLeast(Version.V2) ? MemtableParams.get(in.readUTF()) : MemtableParams.DEFAULT)
- .fastPath(version.isAtLeast(Version.V2) ? FastPathStrategy.serializer.deserialize(in, version) : FastPathStrategy.simple())
.caching(CachingParams.fromMap(deserializeMap(in)))
.compaction(CompactionParams.fromMap(deserializeMap(in)))
.compression(CompressionParams.fromMap(deserializeMap(in)))
.extensions(deserializeMapBB(in))
.cdc(in.readBoolean())
- .readRepair(ReadRepairStrategy.fromString(in.readUTF()))
- .transactionalMode(TransactionalMode.fromOrdinal(in.readInt()))
- .transactionalMigrationFrom(TransactionalMigrationFromMode.fromOrdinal(in.readInt()));
+ .readRepair(ReadRepairStrategy.fromString(in.readUTF()));
+ if (version.isAtLeast(Version.MIN_ACCORD_VERSION))
+ {
+ builder.fastPath(FastPathStrategy.serializer.deserialize(in, version))
+ .transactionalMode(TransactionalMode.fromOrdinal(in.readInt()))
+ .transactionalMigrationFrom(TransactionalMigrationFromMode.fromOrdinal(in.readInt()))
+ .pendingDrop(in.readBoolean());
+ }
return builder.build();
}
public long serializedSize(TableParams t, Version version)
{
- return sizeof(t.comment) +
+ long size = sizeof(t.comment) +
sizeof(t.bloomFilterFpChance) +
sizeof(t.crcCheckChance) +
sizeof(t.gcGraceSeconds) +
@@ -611,15 +631,20 @@ public final class TableParams
sizeof(t.speculativeRetry.toString()) +
sizeof(t.additionalWritePolicy.toString()) +
(version.isAtLeast(Version.V2) ? sizeof(t.memtable.configurationKey()) : 0) +
- (version.isAtLeast(Version.V2) ? FastPathStrategy.serializer.serializedSize(t.fastPath, version) : 0) +
serializedSizeMap(t.caching.asMap()) +
serializedSizeMap(t.compaction.asMap()) +
serializedSizeMap(t.compression.asMap()) +
serializedSizeMapBB(t.extensions) +
sizeof(t.cdc) +
- sizeof(t.readRepair.name()) +
- sizeof(t.transactionalMode.ordinal()) +
- sizeof(t.transactionalMigrationFrom.ordinal());
+ sizeof(t.readRepair.name());
+ if (version.isAtLeast(Version.MIN_ACCORD_VERSION))
+ {
+ size += FastPathStrategy.serializer.serializedSize(t.fastPath, version) +
+ sizeof(t.transactionalMode.ordinal()) +
+ sizeof(t.transactionalMigrationFrom.ordinal()) +
+ sizeof(t.pendingDrop);
+ }
+ return size;
}
private void serializeMap(Map<String, String> map, DataOutputPlus out) throws IOException
diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java
index 5d30a1db57..f41a07a233 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java
@@ -20,9 +20,12 @@ package org.apache.cassandra.service.accord;
import java.util.Map;
import java.util.NavigableMap;
+import java.util.Set;
import java.util.function.BiFunction;
import javax.annotation.Nullable;
+import com.google.common.annotations.VisibleForTesting;
+
import accord.api.Agent;
import accord.api.DataStore;
import accord.api.Key;
@@ -78,6 +81,12 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
return new AccordSafeCommandStore(preLoadContext, commands, timestampsForKey, commandsForKey, commandsForRanges, commandStore);
}
+ @VisibleForTesting
+ public Set<Key> commandsForKeysKeys()
+ {
+ return commandsForKeys.keySet();
+ }
+
@Override
protected AccordSafeCommand getCommandInternal(TxnId txnId)
{
diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java
index cab494dd96..5ecc1d81a4 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordService.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordService.java
@@ -18,6 +18,7 @@
package org.apache.cassandra.service.accord;
+import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -80,13 +81,17 @@ import accord.api.Result;
import accord.coordinate.CoordinationFailed;
import accord.coordinate.Preempted;
import accord.coordinate.Timeout;
+import accord.api.RoutingKey;
+import accord.coordinate.ExecuteSyncPoint;
+import accord.coordinate.tracking.AllTracker;
+import accord.coordinate.tracking.RequestStatus;
import accord.impl.AbstractConfigurationService;
import accord.impl.DefaultLocalListeners;
import accord.impl.DefaultRemoteListeners;
import accord.impl.DefaultRequestTimeouts;
import accord.impl.SizeOfIntersectionSorter;
-import accord.local.CommandStore;
import accord.impl.progresslog.DefaultProgressLogs;
+import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.DurableBefore;
import accord.local.KeyHistory;
@@ -98,7 +103,10 @@ import accord.local.SaveStatus;
import accord.local.ShardDistributor.EvenSplit;
import accord.local.Status;
import accord.local.cfk.CommandsForKey;
+import accord.messages.Callback;
+import accord.messages.ReadData;
import accord.messages.Request;
+import accord.messages.WaitUntilApplied;
import accord.primitives.Keys;
import accord.primitives.Seekable;
import accord.primitives.Seekables;
@@ -106,18 +114,21 @@ import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.Txn.Kind;
import accord.primitives.TxnId;
+import accord.topology.Topologies;
import accord.topology.TopologyManager;
import accord.utils.DefaultRandom;
import accord.utils.Invariants;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
import accord.utils.async.AsyncResult;
+import accord.utils.async.AsyncResults;
import org.agrona.collections.Int2ObjectHashMap;
import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.WriteType;
+import org.apache.cassandra.dht.AccordSplitter;
import org.apache.cassandra.exceptions.ReadTimeoutException;
import org.apache.cassandra.exceptions.RequestTimeoutException;
import org.apache.cassandra.exceptions.WriteTimeoutException;
@@ -126,6 +137,7 @@ import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessageDelivery;
import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.service.accord.AccordSyncPropagator.Notification;
import org.apache.cassandra.service.accord.api.AccordAgent;
@@ -141,8 +153,8 @@ import org.apache.cassandra.service.consensus.TransactionalMode;
import org.apache.cassandra.service.consensus.migration.TableMigrationState;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.membership.NodeId;
-import org.apache.cassandra.transport.Dispatcher;
import org.apache.cassandra.tracing.Tracing;
+import org.apache.cassandra.transport.Dispatcher;
import org.apache.cassandra.utils.Blocking;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.ExecutorUtils;
@@ -428,11 +440,14 @@ public class AccordService implements IAccordService, Shutdownable
{
List<ClusterMetadata> historic = ref.historic;
long lastHistoric = ref.historic.get(historic.size() - 1).epoch.getEpoch();
- // new epochs added while loading... load the deltas
- for (ClusterMetadata metadata : tcmLoadRange(lastHistoric + 1, current.epoch.getEpoch()))
+ if (lastHistoric + 1 < current.epoch.getEpoch())
{
- historic.add(metadata);
- configService.reportMetadataInternal(metadata);
+ // new epochs added while loading... load the deltas
+ for (ClusterMetadata metadata : tcmLoadRange(lastHistoric + 1, current.epoch.getEpoch()))
+ {
+ historic.add(metadata);
+ configService.reportMetadataInternal(metadata);
+ }
}
// sync doesn't happen when this node isn't in the epoch
@@ -506,7 +521,7 @@ public class AccordService implements IAccordService, Shutdownable
List<ClusterMetadata> afterLoad = ClusterMetadataService.instance().processor().reconstructFull(Epoch.create(min - 1), Epoch.create(max));
while (!afterLoad.isEmpty() && afterLoad.get(0).epoch.getEpoch() < min)
afterLoad.remove(0);
- assert !afterLoad.isEmpty() : "TCM was unable to return the needed epochs";
+ assert !afterLoad.isEmpty() : String.format("TCM was unable to return the needed epochs: %d -> %d", min, max);
assert afterLoad.get(0).epoch.getEpoch() == min : String.format("Unexpected epoch: expected %d but given %d", min, afterLoad.get(0).epoch.getEpoch());
assert afterLoad.get(afterLoad.size() - 1).epoch.getEpoch() == max : String.format("Unexpected epoch: expected %d but given %d", max, afterLoad.get(afterLoad.size() - 1).epoch.getEpoch());
return afterLoad;
@@ -1231,4 +1246,227 @@ public class AccordService implements IAccordService, Shutdownable
}));
return new CompactionInfo(redundantBefores, ranges, durableBefore.get());
}
+
+ @Override
+ public void awaitTableDrop(TableId id)
+ {
+ // Need to make sure no existing txn are still being processed for this table... this is only used by DROP TABLE so NEW txn are expected to be blocked, so just need to "wait" for existing ones to complete
+ Topology topology = node.topology().current();
+ List<TokenRange> ranges = topology.reduce(new ArrayList<>(),
+ s -> ((TokenRange) s.range).table().equals(id),
+ (accum, s) -> {
+ accum.add((TokenRange) s.range);
+ return accum;
+ });
+ if (ranges.isEmpty()) return; // nothing to see here
+
+ ColumnFamilyStore cfs = Schema.instance.getColumnFamilyStoreInstance(id);
+ Invariants.checkState(cfs != null, "Unable to find table %s", id);
+ BigInteger targetSplitSize = BigInteger.valueOf(Math.max(1, cfs.estimateKeys() / 1_000_000));
+
+ List<AsyncChain<?>> syncs = new ArrayList<>(ranges.size());
+ for (TokenRange range : ranges)
+ syncs.add(awaitTableDrop(cfs, range, targetSplitSize));
+ AsyncChain<Object[]> all = AsyncChains.allOf(syncs);
+ try
+ {
+ AsyncChains.getBlocking(all);
+ }
+ catch (InterruptedException e)
+ {
+ Thread.currentThread().interrupt();
+ throw new UncheckedInterruptedException(e);
+ }
+ catch (ExecutionException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private AsyncChain<?> awaitTableDrop(ColumnFamilyStore cfs, TokenRange range, BigInteger targetSplitSize)
+ {
+ List<TokenRange> splits = split(cfs, range, targetSplitSize);
+ List<AsyncChain<?>> syncs = new ArrayList<>(splits.size());
+ for (TokenRange tr : splits)
+ syncs.add(awaitTableDropSubRange(tr));
+ return AsyncChains.allOf(syncs);
+ }
+
+ private List<TokenRange> split(ColumnFamilyStore cfs, TokenRange range, BigInteger targetSplitSize)
+ {
+ if (targetSplitSize.equals(BigInteger.ONE)) return Collections.singletonList(range);
+
+ AccordSplitter splitter = cfs.getPartitioner().accordSplitter().apply(Ranges.single(range));
+ RoutingKey remainingStart = range.start();
+
+ BigInteger rangeSize = splitter.sizeOf(range);
+ BigInteger divide = splitter.divide(rangeSize, targetSplitSize);
+ BigInteger rangeStep = divide.equals(BigInteger.ZERO) ? rangeSize : BigInteger.ONE.max(divide);
+ BigInteger offset = BigInteger.ZERO;
+ List<TokenRange> result = new ArrayList<>();
+
+ while (splitter.compare(offset, rangeSize) < 0)
+ {
+ BigInteger remaining = rangeSize.subtract(offset);
+ BigInteger length = remaining.min(rangeStep);
+
+ TokenRange next = splitter.subRange(range, offset, splitter.add(offset, length));
+ result.add(next);
+ remainingStart = next.end();
+ offset = offset.add(length);
+ }
+
+ if (!remainingStart.equals(range.end()))
+ result.add(range.newRange(remainingStart, range.end()));
+ assert result.get(0).start().equals(range.start()) : String.format("Starting range %s does not have the same start as %s", result.get(0), range);
+ assert result.get(result.size() - 1).end().equals(range.end()) : String.format("Ending range %s does not have the same end as %s", result.get(result.size() - 1), range);
+ return result;
+ }
+
+ private AsyncChain<?> awaitTableDropSubRange(TokenRange range)
+ {
+ return awaitTableDropSubRange(Ranges.single(range), 0);
+ }
+
+ private AsyncChain<Void> awaitTableDropSubRange(Ranges ranges, int attempt)
+ {
+ return exclusiveSyncPoint(ranges, attempt)
+ .flatMap(s -> s == null ? AsyncChains.success(null) : Await.coordinate(node, s));
+ }
+
+ private AsyncChain<SyncPoint<Ranges>> exclusiveSyncPoint(Ranges ranges, int attempt)
+ {
+ //TODO (on merge): CASSANDRA-19769 has the same logic... should this be refactored? Would make it nice so we could split the range on retries?
+ return CoordinateSyncPoint.exclusive(node, ranges)
+ .recover(t -> {
+ //TODO (operability): make this configurable / monitorable?
+ if (attempt > 3) return null;
+ switch (shouldRetry(t))
+ {
+ case SUCCESS:
+ return AsyncChains.success(null);
+ case RETRY:
+ return exclusiveSyncPoint(ranges, attempt + 1);
+ case FAIL:
+ return null;
+ default:
+ throw new UnsupportedOperationException();
+ }
+ });
+ }
+
+ private enum RetryDecission { SUCCESS, RETRY, FAIL }
+ private static RetryDecission shouldRetry(Throwable t)
+ {
+ if (t.getClass() == ExecuteSyncPoint.SyncPointErased.class)
+ return RetryDecission.SUCCESS;
+ if (t instanceof Invalidated || t instanceof Preempted || t instanceof Timeout)
+ return RetryDecission.RETRY;
+ return RetryDecission.FAIL;
+ }
+
+ // TODO (duplication): this is 95% of accord.coordinate.CoordinateShardDurable
+ // we already report all this information to EpochState; would be better to use that
+ // Taken from ListStore...
+ private static class Await extends AsyncResults.SettableResult<SyncPoint<Ranges>> implements Callback<ReadData.ReadReply>
+ {
+ private final Node node;
+ private final AllTracker tracker;
+ private final SyncPoint<Ranges> exclusiveSyncPoint;
+
+ private Await(Node node, SyncPoint<Ranges> exclusiveSyncPoint)
+ {
+ Topologies topologies = node.topology().forEpoch(exclusiveSyncPoint.keysOrRanges, exclusiveSyncPoint.sourceEpoch());
+ this.node = node;
+ this.tracker = new AllTracker(topologies);
+ this.exclusiveSyncPoint = exclusiveSyncPoint;
+ }
+
+ public static AsyncChain<Void> coordinate(Node node, SyncPoint<Ranges> sp)
+ {
+ return node.withEpoch(sp.sourceEpoch(), () -> {
+ Await coordinate = new Await(node, sp);
+ coordinate.start();
+ AsyncChain<Void> chain = coordinate.map(i -> null);
+ return chain.recover(t -> {
+ switch (shouldRetry(t))
+ {
+ case SUCCESS: return AsyncChains.success(null);
+ case RETRY: return coordinate(node, sp);
+ case FAIL: return null;
+ default: throw new UnsupportedOperationException();
+ }
+ });
+ });
+ }
+
+ private void start()
+ {
+ node.send(tracker.nodes(), to -> new WaitUntilApplied(to, tracker.topologies(), exclusiveSyncPoint.syncId, exclusiveSyncPoint.keysOrRanges, exclusiveSyncPoint.syncId.epoch()), this);
+ }
+ @Override
+ public void onSuccess(Node.Id from, ReadData.ReadReply reply)
+ {
+ if (!reply.isOk())
+ {
+ ReadData.CommitOrReadNack nack = (ReadData.CommitOrReadNack) reply;
+ switch (nack)
+ {
+ default: throw new AssertionError("Unhandled: " + reply);
+
+ case Insufficient:
+ CoordinateSyncPoint.sendApply(node, from, exclusiveSyncPoint);
+ return;
+ case Rejected:
+ tryFailure(new RuntimeException(nack.name()));
+ case Redundant:
+ tryFailure(new ExecuteSyncPoint.SyncPointErased());
+ return;
+ case Invalid:
+ tryFailure(new Invalidated(exclusiveSyncPoint.syncId, exclusiveSyncPoint.homeKey));
+ return;
+ }
+ }
+ else
+ {
+ if (tracker.recordSuccess(from) == RequestStatus.Success)
+ {
+ node.configService().reportEpochRedundant(exclusiveSyncPoint.keysOrRanges, exclusiveSyncPoint.syncId.epoch());
+ trySuccess(exclusiveSyncPoint);
+ }
+ }
+ }
+
+ private Throwable cause;
+
+ @Override
+ public void onFailure(Node.Id from, Throwable failure)
+ {
+ synchronized (this)
+ {
+ if (cause == null) cause = failure;
+ else
+ {
+ try
+ {
+ cause.addSuppressed(failure);
+ }
+ catch (Throwable t)
+ {
+ // can not always add suppress
+ node.agent().onUncaughtException(failure);
+ }
+ }
+ failure = cause;
+ }
+ if (tracker.recordFailure(from) == RequestStatus.Failed)
+ tryFailure(failure);
+ }
+
+ @Override
+ public void onCallbackFailure(Node.Id from, Throwable failure)
+ {
+ tryFailure(failure);
+ }
+ }
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordStaleReplicas.java b/src/java/org/apache/cassandra/service/accord/AccordStaleReplicas.java
index 2502fa1a63..f729759fe6 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordStaleReplicas.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordStaleReplicas.java
@@ -44,7 +44,7 @@ public class AccordStaleReplicas implements MetadataValue<AccordStaleReplicas>
private final Set<Node.Id> staleIds;
private final Epoch lastModified;
- AccordStaleReplicas(Set<Node.Id> staleIds, Epoch lastModified)
+ public AccordStaleReplicas(Set<Node.Id> staleIds, Epoch lastModified)
{
this.staleIds = staleIds;
this.lastModified = lastModified;
diff --git a/src/java/org/apache/cassandra/service/accord/AccordTopology.java b/src/java/org/apache/cassandra/service/accord/AccordTopology.java
index 45f3b3fd76..eb8f660b0a 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordTopology.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordTopology.java
@@ -65,16 +65,17 @@ public class AccordTopology
private static class ShardLookup extends HashMap<accord.primitives.Range, Shard>
{
- private Shard createOrReuse(accord.primitives.Range range, SortedArrayList<Node.Id> nodes, Set<Node.Id> fastPathElectorate, Set<Node.Id> joining)
+ private Shard createOrReuse(boolean pendingRemoval, accord.primitives.Range range, SortedArrayList<Node.Id> nodes, Set<Node.Id> fastPathElectorate, Set<Node.Id> joining)
{
Shard prev = get(range);
if (prev != null
+ && prev.pendingRemoval == pendingRemoval
&& Objects.equals(prev.nodes, nodes)
&& Objects.equals(prev.fastPathElectorate, fastPathElectorate)
&& Objects.equals(prev.joining, joining))
return prev;
- return new Shard(range, nodes, fastPathElectorate, joining);
+ return new Shard(range, nodes, fastPathElectorate, joining, pendingRemoval);
}
}
@@ -109,7 +110,7 @@ public class AccordTopology
SortedArrayList<Node.Id> fastPath = strategyFor(metadata).calculateFastPath(nodes, unavailable, dcMap);
- return lookup.createOrReuse(tokenRange, nodes, fastPath, pending);
+ return lookup.createOrReuse(metadata.params.pendingDrop, tokenRange, nodes, fastPath, pending);
}
private static KeyspaceShard forRange(KeyspaceMetadata keyspace, Range<Token> range, Directory directory, VersionedEndpoints.ForRange reads, VersionedEndpoints.ForRange writes)
diff --git a/src/java/org/apache/cassandra/service/accord/IAccordService.java b/src/java/org/apache/cassandra/service/accord/IAccordService.java
index be87f1b4a4..a27a29f104 100644
--- a/src/java/org/apache/cassandra/service/accord/IAccordService.java
+++ b/src/java/org/apache/cassandra/service/accord/IAccordService.java
@@ -49,6 +49,7 @@ import org.apache.cassandra.dht.Token;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
+import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.service.accord.api.AccordScheduler;
import org.apache.cassandra.service.accord.txn.TxnResult;
import org.apache.cassandra.tcm.Epoch;
@@ -158,4 +159,8 @@ public interface IAccordService
Long minEpoch(Collection<TokenRange> ranges);
void tryMarkRemoved(Topology topology, Node.Id node);
+ default void awaitTableDrop(TableId id)
+ {
+
+ }
}
diff --git a/src/java/org/apache/cassandra/service/accord/fastpath/InheritKeyspaceFastPathStrategy.java b/src/java/org/apache/cassandra/service/accord/fastpath/InheritKeyspaceFastPathStrategy.java
index 2ffc3c7cbf..39fae55d92 100644
--- a/src/java/org/apache/cassandra/service/accord/fastpath/InheritKeyspaceFastPathStrategy.java
+++ b/src/java/org/apache/cassandra/service/accord/fastpath/InheritKeyspaceFastPathStrategy.java
@@ -28,7 +28,7 @@ import accord.utils.SortedArrays.SortedArrayList;
public class InheritKeyspaceFastPathStrategy implements FastPathStrategy
{
- static final FastPathStrategy instance = new InheritKeyspaceFastPathStrategy();
+ public static final FastPathStrategy instance = new InheritKeyspaceFastPathStrategy();
private static final Map<String, String> SCHEMA_PARAMS = ImmutableMap.of(Kind.KEY, Kind.INHERIT_KEYSPACE.name());
diff --git a/src/java/org/apache/cassandra/service/accord/fastpath/ParameterizedFastPathStrategy.java b/src/java/org/apache/cassandra/service/accord/fastpath/ParameterizedFastPathStrategy.java
index 10828202b6..13eebb8ce8 100644
--- a/src/java/org/apache/cassandra/service/accord/fastpath/ParameterizedFastPathStrategy.java
+++ b/src/java/org/apache/cassandra/service/accord/fastpath/ParameterizedFastPathStrategy.java
@@ -50,8 +50,8 @@ import javax.annotation.Nonnull;
public class ParameterizedFastPathStrategy implements FastPathStrategy
{
- static final String SIZE = "size";
- static final String DCS = "dcs";
+ public static final String SIZE = "size";
+ public static final String DCS = "dcs";
private static final Joiner DC_JOINER = Joiner.on(',');
private static final Pattern COMMA_SEPARATOR = Pattern.compile(",");
private static final Pattern COLON_SEPARATOR = Pattern.compile(":");
@@ -64,7 +64,7 @@ public class ParameterizedFastPathStrategy implements FastPathStrategy
public void serialize(WeightedDc dc, DataOutputPlus out, Version version) throws IOException
{
out.writeUTF(dc.name);
- out.writeUnsignedVInt(dc.weight);
+ out.writeUnsignedVInt32(dc.weight);
out.writeBoolean(dc.autoWeight);
}
@@ -237,7 +237,7 @@ public class ParameterizedFastPathStrategy implements FastPathStrategy
return new ConfigurationException(String.format(fmt, args));
}
- static ParameterizedFastPathStrategy fromMap(Map<String, String> map)
+ public static ParameterizedFastPathStrategy fromMap(Map<String, String> map)
{
if (!map.containsKey(SIZE))
throw cfe("fast_path must be set to 'keyspace' or 'default' or a map defining '%s' and optionally '%s'", SIZE, DCS);
diff --git a/src/java/org/apache/cassandra/service/accord/fastpath/SimpleFastPathStrategy.java b/src/java/org/apache/cassandra/service/accord/fastpath/SimpleFastPathStrategy.java
index 37d51b6c82..8a278faaa2 100644
--- a/src/java/org/apache/cassandra/service/accord/fastpath/SimpleFastPathStrategy.java
+++ b/src/java/org/apache/cassandra/service/accord/fastpath/SimpleFastPathStrategy.java
@@ -31,7 +31,7 @@ import accord.utils.SortedArrays.SortedArrayList;
public class SimpleFastPathStrategy implements FastPathStrategy
{
- static final SimpleFastPathStrategy instance = new SimpleFastPathStrategy();
+ public static final SimpleFastPathStrategy instance = new SimpleFastPathStrategy();
private static final Map<String, String> SCHEMA_PARAMS = ImmutableMap.of(Kind.KEY, Kind.SIMPLE.name());
diff --git a/src/java/org/apache/cassandra/service/consensus/migration/ConsensusMigrationState.java b/src/java/org/apache/cassandra/service/consensus/migration/ConsensusMigrationState.java
index fa6b146d77..731aabb735 100644
--- a/src/java/org/apache/cassandra/service/consensus/migration/ConsensusMigrationState.java
+++ b/src/java/org/apache/cassandra/service/consensus/migration/ConsensusMigrationState.java
@@ -34,6 +34,7 @@ import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Sets;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
@@ -204,6 +205,8 @@ public class ConsensusMigrationState implements MetadataValue<ConsensusMigration
public ConsensusMigrationState withMigrationsRemovedFor(Set<TableId> removed)
{
+ if (tableStates.isEmpty() || Sets.intersection(tableStates.keySet(), removed).isEmpty())
+ return this;
ImmutableMap.Builder<TableId, TableMigrationState> updated = ImmutableMap.builder();
putUnchanged(tableStates, updated, removed);
return new ConsensusMigrationState(lastModified, updated.build());
diff --git a/src/java/org/apache/cassandra/tcm/CMSOperations.java b/src/java/org/apache/cassandra/tcm/CMSOperations.java
index 30ee48b20d..515b71e4f2 100644
--- a/src/java/org/apache/cassandra/tcm/CMSOperations.java
+++ b/src/java/org/apache/cassandra/tcm/CMSOperations.java
@@ -33,6 +33,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.locator.CMSPlacementStrategy;
import org.apache.cassandra.schema.ReplicationParams;
+import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.membership.NodeState;
import org.apache.cassandra.tcm.membership.NodeVersion;
@@ -40,6 +41,7 @@ import org.apache.cassandra.tcm.sequences.CancelCMSReconfiguration;
import org.apache.cassandra.tcm.sequences.InProgressSequences;
import org.apache.cassandra.tcm.sequences.ReconfigureCMS;
import org.apache.cassandra.tcm.serialization.Version;
+import org.apache.cassandra.tcm.sequences.DropAccordTable;
import org.apache.cassandra.tcm.transformations.Unregister;
import org.apache.cassandra.tcm.transformations.cms.AdvanceCMSReconfiguration;
import org.apache.cassandra.utils.FBUtilities;
@@ -255,4 +257,19 @@ public class CMSOperations implements CMSOperationsMBean
cms.commit(new Unregister(nodeId, EnumSet.of(NodeState.LEFT)));
}
}
+
+ @Override
+ public void resumeDropAccordTable(String tableId)
+ {
+ TableId id = TableId.fromString(tableId);
+ for (MultiStepOperation.SequenceKey key : ClusterMetadata.current().inProgressSequences.keys())
+ {
+ if (key instanceof DropAccordTable.TableReference && ((DropAccordTable.TableReference) key).id.equals(id))
+ {
+ InProgressSequences.finishInProgressSequences(key);
+ return;
+ }
+ }
+ throw new IllegalArgumentException("No drop table operation is in progress for table with id " + tableId);
+ }
}
diff --git a/src/java/org/apache/cassandra/tcm/CMSOperationsMBean.java b/src/java/org/apache/cassandra/tcm/CMSOperationsMBean.java
index 3b6e4b526b..9fd7539e40 100644
--- a/src/java/org/apache/cassandra/tcm/CMSOperationsMBean.java
+++ b/src/java/org/apache/cassandra/tcm/CMSOperationsMBean.java
@@ -45,4 +45,5 @@ public interface CMSOperationsMBean
public boolean cancelInProgressSequences(String sequenceOwner, String expectedSequenceKind);
public void unregisterLeftNodes(List<String> nodeIds);
+ public void resumeDropAccordTable(String tableId);
}
diff --git a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java
index c4896de055..809ed7356a 100644
--- a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java
+++ b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java
@@ -81,7 +81,7 @@ import org.apache.cassandra.utils.Pair;
import static com.google.common.collect.ImmutableSet.toImmutableSet;
import static org.apache.cassandra.config.CassandraRelevantProperties.LINE_SEPARATOR;
import static org.apache.cassandra.db.TypeSizes.sizeof;
-import static org.apache.cassandra.tcm.serialization.Version.V2;
+import static org.apache.cassandra.tcm.serialization.Version.MIN_ACCORD_VERSION;
public class ClusterMetadata
{
@@ -199,16 +199,6 @@ public class ClusterMetadata
this.accordStaleReplicas = accordStaleReplicas;
}
- public ClusterMetadata withDirectory(Directory directory)
- {
- return new ClusterMetadata(epoch, partitioner, schema, directory, tokenMap, placements, accordFastPath, lockedRanges, inProgressSequences, consensusMigrationState, extensions, accordStaleReplicas);
- }
-
- public ClusterMetadata withPlacements(DataPlacements placements)
- {
- return new ClusterMetadata(epoch, partitioner, schema, directory, tokenMap, placements, accordFastPath, lockedRanges, inProgressSequences, consensusMigrationState, extensions, accordStaleReplicas);
- }
-
public Set<InetAddressAndPort> fullCMSMembers()
{
if (fullCMSEndpoints == null)
@@ -321,7 +311,7 @@ public class ClusterMetadata
while (iter.hasNext())
{
Transformation.Result result = iter.next().applyTo(metadata);
- assert result.isSuccess();
+ assert result.isSuccess() : result.toString();
metadata = result.success().metadata;
}
return metadata.placements.get(ksm.params.replication);
@@ -620,7 +610,7 @@ public class ClusterMetadata
public Transformer with(ExtensionKey<?, ?> key, ExtensionValue<?> obj)
{
- if (MetadataKeys.CORE_METADATA.contains(key))
+ if (MetadataKeys.CORE_METADATA.containsKey(key))
throw new IllegalArgumentException("Core cluster metadata objects should be addressed directly, " +
"not using the associated MetadataKey");
@@ -643,7 +633,7 @@ public class ClusterMetadata
public Transformer without(ExtensionKey<?, ?> key)
{
- if (MetadataKeys.CORE_METADATA.contains(key))
+ if (MetadataKeys.CORE_METADATA.containsKey(key))
throw new IllegalArgumentException("Core cluster metadata objects should be addressed directly, " +
"not using the associated MetadataKey");
if (extensions.remove(key) != null)
@@ -742,7 +732,7 @@ public class ClusterMetadata
inProgressSequences,
consensusMigrationState,
extensions,
- accordStaleReplicas),
+ accordStaleReplicas),
ImmutableSet.copyOf(modifiedKeys));
}
@@ -1018,7 +1008,7 @@ public class ClusterMetadata
Directory.serializer.serialize(metadata.directory, out, version);
TokenMap.serializer.serialize(metadata.tokenMap, out, version);
DataPlacements.serializer.serialize(metadata.placements, out, version);
- if (version.isAtLeast(V2))
+ if (version.isAtLeast(MIN_ACCORD_VERSION))
{
AccordFastPath.serializer.serialize(metadata.accordFastPath, out, version);
ConsensusMigrationState.serializer.serialize(metadata.consensusMigrationState, out, version);
@@ -1066,7 +1056,7 @@ public class ClusterMetadata
ConsensusMigrationState consensusMigrationState;
AccordStaleReplicas staleReplicas;
- if (version.isAtLeast(V2))
+ if (version.isAtLeast(MIN_ACCORD_VERSION))
{
accordFastPath = AccordFastPath.serializer.deserialize(in, version);
consensusMigrationState = ConsensusMigrationState.serializer.deserialize(in, version);
@@ -1123,7 +1113,7 @@ public class ClusterMetadata
TokenMap.serializer.serializedSize(metadata.tokenMap, version) +
DataPlacements.serializer.serializedSize(metadata.placements, version);
- if (version.isAtLeast(V2))
+ if (version.isAtLeast(MIN_ACCORD_VERSION))
{
size += AccordFastPath.serializer.serializedSize(metadata.accordFastPath, version) +
ConsensusMigrationState.serializer.serializedSize(metadata.consensusMigrationState, version) +
diff --git a/src/java/org/apache/cassandra/tcm/MetadataKeys.java b/src/java/org/apache/cassandra/tcm/MetadataKeys.java
index 62bd72fe70..88f5dcb3f2 100644
--- a/src/java/org/apache/cassandra/tcm/MetadataKeys.java
+++ b/src/java/org/apache/cassandra/tcm/MetadataKeys.java
@@ -24,6 +24,7 @@ import java.util.Map;
import java.util.Set;
import java.util.function.Function;
+import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import org.apache.cassandra.tcm.extensions.ExtensionKey;
@@ -43,15 +44,18 @@ public class MetadataKeys
public static final MetadataKey IN_PROGRESS_SEQUENCES = make(CORE_NS, "sequences", "in_progress");
public static final MetadataKey CONSENSUS_MIGRATION_STATE = make(CORE_NS, "consensus", "migration_state");
- public static final ImmutableSet<MetadataKey> CORE_METADATA = ImmutableSet.of(SCHEMA,
- NODE_DIRECTORY,
- TOKEN_MAP,
- DATA_PLACEMENTS,
- ACCORD_FAST_PATH,
- ACCORD_STALE_REPLICAS,
- LOCKED_RANGES,
- IN_PROGRESS_SEQUENCES,
- CONSENSUS_MIGRATION_STATE);
+ public static final ImmutableMap<MetadataKey, Function<ClusterMetadata, MetadataValue<?>>> CORE_METADATA
+ = ImmutableMap.<MetadataKey, Function<ClusterMetadata, MetadataValue<?>>>builder()
+ .put(SCHEMA, cm -> cm.schema)
+ .put(NODE_DIRECTORY, cm -> cm.directory)
+ .put(TOKEN_MAP, cm -> cm.tokenMap)
+ .put(DATA_PLACEMENTS, cm -> cm.placements)
+ .put(LOCKED_RANGES, cm -> cm.lockedRanges)
+ .put(IN_PROGRESS_SEQUENCES, cm -> cm.inProgressSequences)
+ .put(ACCORD_FAST_PATH, cm -> cm.accordFastPath)
+ .put(ACCORD_STALE_REPLICAS, cm -> cm.accordStaleReplicas)
+ .put(CONSENSUS_MIGRATION_STATE, cm -> cm.consensusMigrationState)
+ .build();
public static MetadataKey make(String...parts)
{
@@ -65,6 +69,15 @@ public class MetadataKeys
return new MetadataKey(b.toString());
}
+ public static MetadataValue<?> extract(ClusterMetadata cm, MetadataKey key)
+ {
+ if (CORE_METADATA.containsKey(key))
+ return CORE_METADATA.get(key).apply(cm);
+ if (!(key instanceof ExtensionKey<?, ?>))
+ throw new IllegalArgumentException("Unknown key: " + key);
+ return cm.extensions.get(key);
+ }
+
public static ImmutableSet<MetadataKey> diffKeys(ClusterMetadata before, ClusterMetadata after)
{
ImmutableSet.Builder<MetadataKey> builder = new ImmutableSet.Builder<>();
@@ -74,12 +87,8 @@ public class MetadataKeys
private static void diffKeys(ClusterMetadata before, ClusterMetadata after, ImmutableSet.Builder<MetadataKey> builder)
{
- checkKey(before, after, builder, cm -> cm.schema, MetadataKeys.SCHEMA);
- checkKey(before, after, builder, cm -> cm.directory, MetadataKeys.NODE_DIRECTORY);
- checkKey(before, after, builder, cm -> cm.tokenMap, MetadataKeys.TOKEN_MAP);
- checkKey(before, after, builder, cm -> cm.placements, MetadataKeys.DATA_PLACEMENTS);
- checkKey(before, after, builder, cm -> cm.lockedRanges, MetadataKeys.LOCKED_RANGES);
- checkKey(before, after, builder, cm -> cm.inProgressSequences, MetadataKeys.IN_PROGRESS_SEQUENCES);
+ for (Map.Entry<MetadataKey, Function<ClusterMetadata, MetadataValue<?>>> e : CORE_METADATA.entrySet())
+ checkKey(before, after, builder, e.getValue(), e.getKey());
Set<ExtensionKey<?,?>> added = new HashSet<>(after.extensions.keySet());
for (Map.Entry<ExtensionKey<?, ?>, ExtensionValue<?>> entry : before.extensions.entrySet())
diff --git a/src/java/org/apache/cassandra/tcm/MultiStepOperation.java b/src/java/org/apache/cassandra/tcm/MultiStepOperation.java
index 5fffbd2c84..cf634de51b 100644
--- a/src/java/org/apache/cassandra/tcm/MultiStepOperation.java
+++ b/src/java/org/apache/cassandra/tcm/MultiStepOperation.java
@@ -33,6 +33,7 @@ import org.apache.cassandra.tcm.sequences.ProgressBarrier;
import org.apache.cassandra.tcm.sequences.UnbootstrapAndLeave;
import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
import org.apache.cassandra.tcm.serialization.MetadataSerializer;
+import org.apache.cassandra.tcm.sequences.DropAccordTable;
/**
* Represents a multi-step process performed in order to transition the cluster to some state.
@@ -67,7 +68,8 @@ public abstract class MultiStepOperation<CONTEXT>
LEAVE(UnbootstrapAndLeave.serializer),
REMOVE(UnbootstrapAndLeave.serializer),
- RECONFIGURE_CMS(ReconfigureCMS.serializer)
+ RECONFIGURE_CMS(ReconfigureCMS.serializer),
+ DROP_ACCORD_TABLE(DropAccordTable.serializer),
;
public final AsymmetricMetadataSerializer<MultiStepOperation<?>, ? extends MultiStepOperation<?>> serializer;
diff --git a/src/java/org/apache/cassandra/tcm/StubClusterMetadataService.java b/src/java/org/apache/cassandra/tcm/StubClusterMetadataService.java
index 42924d1690..47a2ba26f1 100644
--- a/src/java/org/apache/cassandra/tcm/StubClusterMetadataService.java
+++ b/src/java/org/apache/cassandra/tcm/StubClusterMetadataService.java
@@ -76,7 +76,7 @@ public class StubClusterMetadataService extends ClusterMetadataService
private ClusterMetadata metadata;
- private StubClusterMetadataService(ClusterMetadata initial)
+ protected StubClusterMetadataService(ClusterMetadata initial)
{
super(new UniformRangePlacement(),
MetadataSnapshots.NO_OP,
@@ -107,15 +107,20 @@ public class StubClusterMetadataService extends ClusterMetadataService
@Override
public <T1> T1 commit(Transformation transform, CommitSuccessHandler<T1> onSuccess, CommitFailureHandler<T1> onFailure)
{
- Transformation.Result result = transform.execute(metadata);
+ Transformation.Result result = execute(transform);
if (result.isSuccess())
{
- metadata = result.success().metadata;
+ setMetadata(result.success().metadata);
return onSuccess.accept(result.success().metadata);
}
return onFailure.accept(result.rejected().code, result.rejected().reason);
}
+ protected Transformation.Result execute(Transformation transform)
+ {
+ return transform.execute(metadata());
+ }
+
@Override
public ClusterMetadata fetchLogFromCMS(Epoch awaitAtLeast)
{
diff --git a/src/java/org/apache/cassandra/tcm/Transformation.java b/src/java/org/apache/cassandra/tcm/Transformation.java
index a66ddae050..fb0181568b 100644
--- a/src/java/org/apache/cassandra/tcm/Transformation.java
+++ b/src/java/org/apache/cassandra/tcm/Transformation.java
@@ -45,8 +45,10 @@ import org.apache.cassandra.tcm.transformations.Assassinate;
import org.apache.cassandra.tcm.transformations.BeginConsensusMigrationForTableAndRange;
import org.apache.cassandra.tcm.transformations.CancelInProgressSequence;
import org.apache.cassandra.tcm.transformations.CustomTransformation;
+import org.apache.cassandra.tcm.transformations.FinishDropAccordTable;
import org.apache.cassandra.tcm.transformations.ForceSnapshot;
import org.apache.cassandra.tcm.transformations.MaybeFinishConsensusMigrationForTableAndRange;
+import org.apache.cassandra.tcm.transformations.PrepareDropAccordTable;
import org.apache.cassandra.tcm.transformations.PrepareJoin;
import org.apache.cassandra.tcm.transformations.PrepareLeave;
import org.apache.cassandra.tcm.transformations.PrepareMove;
@@ -238,7 +240,8 @@ public interface Transformation
MAYBE_FINISH_CONSENSUS_MIGRATION_FOR_TABLE_AND_RANGE(37, () -> MaybeFinishConsensusMigrationForTableAndRange.serializer),
ACCORD_MARK_STALE(38, () -> AccordMarkStale.serializer),
ACCORD_MARK_REJOINING(39, () -> AccordMarkRejoining.serializer),
-
+ PREPARE_DROP_ACCORD_TABLE(40, () -> PrepareDropAccordTable.serializer),
+ FINISH_DROP_ACCORD_TABLE(41, () -> FinishDropAccordTable.serializer),
;
private final Supplier<AsymmetricMetadataSerializer<Transformation, ? extends Transformation>> serializer;
diff --git a/src/java/org/apache/cassandra/tcm/membership/NodeVersion.java b/src/java/org/apache/cassandra/tcm/membership/NodeVersion.java
index 4428a169e3..2d8523adfb 100644
--- a/src/java/org/apache/cassandra/tcm/membership/NodeVersion.java
+++ b/src/java/org/apache/cassandra/tcm/membership/NodeVersion.java
@@ -34,7 +34,7 @@ import static org.apache.cassandra.db.TypeSizes.sizeofUnsignedVInt;
public class NodeVersion implements Comparable<NodeVersion>
{
public static final Serializer serializer = new Serializer();
- public static final Version CURRENT_METADATA_VERSION = Version.V2;
+ public static final Version CURRENT_METADATA_VERSION = Version.V3;
public static final NodeVersion CURRENT = new NodeVersion(new CassandraVersion(FBUtilities.getReleaseVersionString()), CURRENT_METADATA_VERSION);
private static final CassandraVersion SINCE_VERSION = CassandraVersion.CASSANDRA_5_0;
diff --git a/src/java/org/apache/cassandra/tcm/sequences/DropAccordTable.java b/src/java/org/apache/cassandra/tcm/sequences/DropAccordTable.java
new file mode 100644
index 0000000000..46047ef494
--- /dev/null
+++ b/src/java/org/apache/cassandra/tcm/sequences/DropAccordTable.java
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tcm.sequences;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.schema.Keyspaces;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.accord.AccordService;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.ClusterMetadataService;
+import org.apache.cassandra.tcm.Epoch;
+import org.apache.cassandra.tcm.MultiStepOperation;
+import org.apache.cassandra.tcm.Transformation;
+import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
+import org.apache.cassandra.tcm.serialization.MetadataSerializer;
+import org.apache.cassandra.tcm.serialization.Version;
+import org.apache.cassandra.tcm.transformations.FinishDropAccordTable;
+import org.apache.cassandra.utils.JVMStabilityInspector;
+
+import static org.apache.cassandra.tcm.Transformation.Kind.FINISH_DROP_ACCORD_TABLE;
+import static org.apache.cassandra.tcm.sequences.SequenceState.continuable;
+import static org.apache.cassandra.tcm.sequences.SequenceState.error;
+import static org.apache.cassandra.tcm.sequences.SequenceState.halted;
+import static org.apache.cassandra.utils.Clock.Global.nanoTime;
+
+/**
+ * A slightly atypical implementation as it consists of only a single step. To perform the drop of an
+ * Accord table, we first commit a PrepareDropAccordTable transformation. Upon enactement, that
+ * marks the table as pending drop, which blocks any new transactions from being started. It also
+ * instantiates an instance of this operation and adds it to the set of in progress operations.
+ *
+ * The intention is to introduce a barrier which blocks until the Accord service acknowledges that
+ * it was learned of the epoch in which the table was marked for deletion and that all prior transactions
+ * are completed. Once this is complete, we can proceed to actually drop the table. The transformation
+ * which performs that schema modification also removes this MSO from ClusterMetadata's in-flight set.
+ * This obviates the need to 'advance' this MSO in the way that other implementations with more steps do.
+ *
+ */
+public class DropAccordTable extends MultiStepOperation<Epoch>
+{
+ private static final Logger logger = LoggerFactory.getLogger(DropAccordTable.class);
+
+ public static final Serializer serializer = new Serializer();
+
+ public final TableReference table;
+
+ public static DropAccordTable newSequence(TableReference table, Epoch preparedAt)
+ {
+ return new DropAccordTable(table, preparedAt);
+ }
+
+ /**
+ * Used by factory method for external callers and by the serializer.
+ * We don't need to include the serialized FinishDropAccordTable step in the serialization
+ * of the MSO itself because they have no parameters other than the table reference and so
+ * we can just construct a new one when we execute it
+ */
+ private DropAccordTable(TableReference table, Epoch latestModification)
+ {
+ super(0, latestModification);
+ this.table = table;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ DropAccordTable that = (DropAccordTable) o;
+ return latestModification.equals(that.latestModification)
+ && table.equals(that.table);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(latestModification, table);
+ }
+
+ @Override
+ public Kind kind()
+ {
+ return Kind.DROP_ACCORD_TABLE;
+ }
+
+ @Override
+ protected SequenceKey sequenceKey()
+ {
+ return table;
+ }
+
+ @Override
+ public MetadataSerializer<? extends SequenceKey> keySerializer()
+ {
+ return TableReference.serializer;
+ }
+
+ @Override
+ public Transformation.Kind nextStep()
+ {
+ return FINISH_DROP_ACCORD_TABLE;
+ }
+
+ @Override
+ public SequenceState executeNext()
+ {
+ try
+ {
+ SequenceState failure = awaitSafeFromAccord();
+ if (failure != null) return failure;
+ }
+ catch (Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t);
+ logger.warn("Exception while waiting for Accord service to notify all table txns are complete", t);
+ // this is actually continuable as we can simply retry
+ return continuable();
+ }
+ try
+ {
+ // Now we're satisfied that all Accord txns have finished for the table,
+ // go ahead and actually drop it
+ ClusterMetadataService.instance().commit(new FinishDropAccordTable(table));
+ return continuable();
+ }
+ catch (Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t);
+ logger.warn("Exception committing finish_drop_accord_table. " +
+ "Accord service has acknowledged the operation but table remains present in schema", t);
+ return halted();
+ }
+ }
+
+ private SequenceState awaitSafeFromAccord() throws ExecutionException, InterruptedException
+ {
+ // make sure that Accord sees the current epoch, which must necessarily follow the
+ // one which marked the table as pending drop
+ ClusterMetadata metadata = ClusterMetadata.current();
+ // just for the sake of paranoia, assert that the table is actually marked as being dropped
+ if (!verifyTableMarked(metadata.schema.getKeyspaces()))
+ return error(new IllegalStateException(String.format("Table %s is in an invalid state to be dropped", table)));
+
+ long startNanos = nanoTime();
+ AccordService.instance().epochReady(metadata.epoch).get();
+ long epochEndNanos = nanoTime();
+
+ // As of this writing this logic is based off ExclusiveSyncPoints which is a bit heavy weight for what is needed, this could cause timeouts for clusters that have a lot of data.
+ // There are retries baked into this call, but trying to handle timeouts more broadly is put on hold as there is active work to define a EpochSyncPoint that should be far cheaper
+ // which would avoid the timeout issues
+ // NOTE: ExclusiveSyncPoint must find all keys in the range, then make sure nothing is blocking them... this causes a lot of IO. EpochSyncPoint just needs to validate that the last txn processed is in the newer epoch, this can work with in-memory state.
+ AccordService.instance().awaitTableDrop(table.id);
+ long awaitEndNanos = nanoTime();
+ logger.info("Wait for Accord to see the drop table was success. " +
+ "Took {} to wait for Accord to learn about the change, then {} to process everything",
+ Duration.ofNanos(epochEndNanos - startNanos), Duration.ofNanos(awaitEndNanos - epochEndNanos));
+ return null;
+ }
+
+ private boolean verifyTableMarked(Keyspaces keyspaces)
+ {
+ TableMetadata tm = keyspaces.getTableOrViewNullable(table.id);
+ if (tm == null)
+ {
+ logger.warn("Unable to drop accord table {}, table not found", table);
+ return false;
+ }
+
+ if (!tm.params.pendingDrop)
+ {
+ logger.warn("Unexpected state, table {} was not marked pending drop", table);
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public Transformation.Result applyTo(ClusterMetadata metadata)
+ {
+ // note: that this will apply the finish drop transformation to the supplied metadata. It's
+ // not used to actually execute the MSO, but to determine what the metadata state will/would
+ // be if it were executed.
+ return new FinishDropAccordTable(table).execute(metadata);
+ }
+
+ @Override
+ public DropAccordTable advance(Epoch epoch)
+ {
+ // note: this isn't really used by this MSO impl as it consists of a single step so there's nothing
+ // to advance. An action of the single step is to remove the MSO from the set of in progress sequences
+ return new DropAccordTable(this.table, epoch);
+ }
+
+ @Override
+ public ProgressBarrier barrier()
+ {
+ return ProgressBarrier.immediate();
+ }
+
+ public static class TableReference implements SequenceKey, Comparable<TableReference>
+ {
+ public static final Serializer serializer = new Serializer();
+
+ public final TableId id;
+
+ public TableReference(TableId id)
+ {
+ this.id = id;
+ }
+
+ public static TableReference from(TableMetadata metadata)
+ {
+ return new TableReference(metadata.id);
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ TableReference that = (TableReference) o;
+ return id.equals(that.id);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(id);
+ }
+
+ @Override
+ public int compareTo(TableReference o)
+ {
+ return id.compareTo(o.id);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "TableReference{id=" + id + '}';
+ }
+
+ public static class Serializer implements MetadataSerializer<TableReference>
+ {
+ @Override
+ public void serialize(TableReference t, DataOutputPlus out, Version version) throws IOException
+ {
+ t.id.serialize(out);
+ }
+
+ @Override
+ public TableReference deserialize(DataInputPlus in, Version version) throws IOException
+ {
+ TableId id = TableId.deserialize(in);
+ return new TableReference(id);
+ }
+
+ @Override
+ public long serializedSize(TableReference t, Version version)
+ {
+ return t.id.serializedSize();
+ }
+ }
+ }
+
+ public static class Serializer implements AsymmetricMetadataSerializer<MultiStepOperation<?>, DropAccordTable>
+ {
+ @Override
+ public void serialize(MultiStepOperation<?> t, DataOutputPlus out, Version version) throws IOException
+ {
+ DropAccordTable plan = (DropAccordTable) t;
+ Epoch.serializer.serialize(plan.latestModification, out, version);
+ // This type of sequence only has a single step so no need to include the index in serde.
+ // Similarly, the only parameter to that single step (FinishDropAccordTable) is the table
+ // reference, so that's all we really need to include in the serialization.
+ TableReference.serializer.serialize(plan.table, out, version);
+ }
+
+ @Override
+ public DropAccordTable deserialize(DataInputPlus in, Version version) throws IOException
+ {
+ Epoch lastModified = Epoch.serializer.deserialize(in, version);
+ TableReference table = TableReference.serializer.deserialize(in, version);
+ return new DropAccordTable(table, lastModified);
+ }
+
+ @Override
+ public long serializedSize(MultiStepOperation<?> t, Version version)
+ {
+ DropAccordTable plan = (DropAccordTable) t;
+ long size = 0;
+ size += Epoch.serializer.serializedSize(plan.latestModification, version);
+ size += TableReference.serializer.serializedSize(plan.table, version);
+ return size;
+ }
+ }
+}
diff --git a/src/java/org/apache/cassandra/tcm/sequences/InProgressSequences.java b/src/java/org/apache/cassandra/tcm/sequences/InProgressSequences.java
index 86a8ec019a..5a94f1826d 100644
--- a/src/java/org/apache/cassandra/tcm/sequences/InProgressSequences.java
+++ b/src/java/org/apache/cassandra/tcm/sequences/InProgressSequences.java
@@ -27,6 +27,7 @@ import java.util.function.Function;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
@@ -59,20 +60,20 @@ public class InProgressSequences implements MetadataValue<InProgressSequences>,
this.state = state;
}
- public static void finishInProgressSequences(MultiStepOperation.SequenceKey sequenceKey)
+ public static ClusterMetadata finishInProgressSequences(MultiStepOperation.SequenceKey sequenceKey)
{
ClusterMetadata metadata = ClusterMetadata.current();
while (true)
{
MultiStepOperation<?> sequence = metadata.inProgressSequences.get(sequenceKey);
if (sequence == null)
- break;
+ return metadata;
if (isLeave(sequence))
StorageService.instance.maybeInitializeServices();
if (resume(sequence))
metadata = ClusterMetadata.current();
else
- return;
+ return metadata;
}
}
@@ -218,6 +219,11 @@ public class InProgressSequences implements MetadataValue<InProgressSequences>,
return state.values().iterator();
}
+ public ImmutableSet<MultiStepOperation.SequenceKey> keys()
+ {
+ return state.keySet();
+ }
+
public static class Serializer implements MetadataSerializer<InProgressSequences>
{
public void serialize(InProgressSequences t, DataOutputPlus out, Version version) throws IOException
diff --git a/src/java/org/apache/cassandra/tcm/serialization/Version.java b/src/java/org/apache/cassandra/tcm/serialization/Version.java
index e4021ecea0..add51eb944 100644
--- a/src/java/org/apache/cassandra/tcm/serialization/Version.java
+++ b/src/java/org/apache/cassandra/tcm/serialization/Version.java
@@ -18,7 +18,10 @@
package org.apache.cassandra.tcm.serialization;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import org.apache.cassandra.tcm.ClusterMetadata;
@@ -36,13 +39,24 @@ public enum Version
/**
* - Added version to PlacementForRange serializer
* - Serialize MemtableParams when serializing TableParams
+ */
+ V2(2),
+
+ /**
* - Added AccordFastPath
+ * - Added ConsensusMigrationState
* - Added AccordStaleReplicas
+ * - TableParam now has pendingDrop (accord table drop is multistep)
*/
- V2(2),
+ V3(3),
UNKNOWN(Integer.MAX_VALUE);
+ /**
+ * The version that Accord was added to TCM.
+ */
+ public static final Version MIN_ACCORD_VERSION = V3;
+
private static Map<Integer, Version> values = new HashMap<>();
static
{
@@ -96,4 +110,15 @@ public enum Version
throw new IllegalArgumentException("Unsupported metadata version (" + i + ")");
}
+
+ public List<Version> greaterThanOrEqual()
+ {
+ Version[] all = Version.values();
+ if (ordinal() == all.length - 1)
+ return Collections.singletonList(this);
+ List<Version> values = new ArrayList<>(all.length - ordinal());
+ for (int i = ordinal(); i < all.length; i++)
+ values.add(all[i]);
+ return values;
+ }
}
diff --git a/src/java/org/apache/cassandra/tcm/transformations/AlterSchema.java b/src/java/org/apache/cassandra/tcm/transformations/AlterSchema.java
index 8c7604d792..c97c6e3be2 100644
--- a/src/java/org/apache/cassandra/tcm/transformations/AlterSchema.java
+++ b/src/java/org/apache/cassandra/tcm/transformations/AlterSchema.java
@@ -248,7 +248,7 @@ public class AlterSchema implements Transformation
return byReplication;
}
- private Transformer maybeUpdateConsensusMigrationState(ConsensusMigrationState prev, Transformer next, ImmutableList<KeyspaceDiff> altered, Keyspaces dropped)
+ public static Transformer maybeUpdateConsensusMigrationState(ConsensusMigrationState prev, Transformer next, ImmutableList<KeyspaceDiff> altered, Keyspaces dropped)
{
ConsensusMigrationState migrationState = prev;
diff --git a/src/java/org/apache/cassandra/tcm/transformations/FinishDropAccordTable.java b/src/java/org/apache/cassandra/tcm/transformations/FinishDropAccordTable.java
new file mode 100644
index 0000000000..2019adb8ad
--- /dev/null
+++ b/src/java/org/apache/cassandra/tcm/transformations/FinishDropAccordTable.java
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tcm.transformations;
+
+import java.io.IOException;
+import java.util.Objects;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.schema.DistributedSchema;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.schema.Keyspaces;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.Transformation;
+import org.apache.cassandra.tcm.sequences.DropAccordTable.TableReference;
+import org.apache.cassandra.tcm.sequences.LockedRanges;
+import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
+import org.apache.cassandra.tcm.serialization.Version;
+
+import static org.apache.cassandra.tcm.Transformation.Kind.FINISH_DROP_ACCORD_TABLE;
+
+/**
+ * Dropping an Accord table is a three-step process.
+ * <ol>
+ * <li>Mark the table as pending drop</li>
+ * <li>Await all in-flight txns to finish</li>
+ * <li>Drop the table from schema (this step)</li>
+ * </ol>
+ * <p/>
+ * Hypothetically it is possible that after {1} has been committed, but before {3} is executed
+ * interleaving metadata changes occur. These could include dropping the table's keyspace, or
+ * modifying the transactional mode of the table to make it a non-accord table. Validation
+ * exists to prevent these schema changes from being committed while the drop is in-flight.
+ * However, if something like this did happen, by the time we come to execute this transformation,
+ * there's nothing really to do other than return success (as the table has indeed already been dropped).
+ */
+public class FinishDropAccordTable implements Transformation
+{
+ private static final Logger logger = LoggerFactory.getLogger(FinishDropAccordTable.class);
+
+ public static final Serializer serializer = new Serializer();
+ public final TableReference tableRef;
+
+ public FinishDropAccordTable(TableReference tableRef)
+ {
+ this.tableRef = tableRef;
+ }
+
+ @Override
+ public Kind kind()
+ {
+ return FINISH_DROP_ACCORD_TABLE;
+ }
+
+ @Override
+ public Result execute(ClusterMetadata prev)
+ {
+ // In every case we remove the operation to drop this table from the set of in-flight sequences
+ ClusterMetadata.Transformer proposed = prev.transformer()
+ .with(prev.inProgressSequences.without(tableRef));
+
+ Keyspaces keyspaces = prev.schema.getKeyspaces();
+ TableMetadata table = keyspaces.getTableOrViewNullable(tableRef.id);
+ // Table was already dropped
+ if (table == null)
+ {
+ logger.warn("Table {} was dropped while drop accord table sequence was in flight", tableRef);
+ return Transformation.success(proposed, LockedRanges.AffectedRanges.EMPTY);
+ }
+ KeyspaceMetadata keyspace = keyspaces.getNullable(table.keyspace);
+
+ // Actually drop the table
+ Keyspaces withoutTable = keyspaces.withAddedOrUpdated(keyspace.withSwapped(keyspace.tables.without(table)));
+
+ Keyspaces.KeyspacesDiff diff = Keyspaces.diff(prev.schema.getKeyspaces(), withoutTable);
+
+ proposed = AlterSchema.maybeUpdateConsensusMigrationState(prev.consensusMigrationState, proposed, diff.altered, Keyspaces.NONE);
+
+ proposed = proposed.with(new DistributedSchema(withoutTable));
+ return Transformation.success(proposed, LockedRanges.AffectedRanges.EMPTY);
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) return true;
+ if (!(o instanceof FinishDropAccordTable)) return false;
+
+ FinishDropAccordTable that = (FinishDropAccordTable) o;
+
+ return Objects.equals(tableRef, that.tableRef);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(tableRef);
+ }
+
+ public static class Serializer implements AsymmetricMetadataSerializer<Transformation, FinishDropAccordTable>
+ {
+ @Override
+ public void serialize(Transformation t, DataOutputPlus out, Version version) throws IOException
+ {
+ FinishDropAccordTable plan = (FinishDropAccordTable) t;
+ TableReference.serializer.serialize(plan.tableRef, out, version);
+ }
+
+ @Override
+ public FinishDropAccordTable deserialize(DataInputPlus in, Version version) throws IOException
+ {
+ TableReference table = TableReference.serializer.deserialize(in, version);
+ return new FinishDropAccordTable(table);
+ }
+
+ @Override
+ public long serializedSize(Transformation t, Version version)
+ {
+ FinishDropAccordTable plan = (FinishDropAccordTable) t;
+ return TableReference.serializer.serializedSize(plan.tableRef, version);
+ }
+ }
+}
\ No newline at end of file
diff --git a/src/java/org/apache/cassandra/tcm/transformations/PrepareDropAccordTable.java b/src/java/org/apache/cassandra/tcm/transformations/PrepareDropAccordTable.java
new file mode 100644
index 0000000000..2c960a0bd3
--- /dev/null
+++ b/src/java/org/apache/cassandra/tcm/transformations/PrepareDropAccordTable.java
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tcm.transformations;
+
+import java.io.IOException;
+import java.util.Objects;
+
+import org.apache.cassandra.exceptions.ExceptionCode;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.schema.DistributedSchema;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.Transformation;
+import org.apache.cassandra.tcm.sequences.DropAccordTable;
+import org.apache.cassandra.tcm.sequences.DropAccordTable.TableReference;
+import org.apache.cassandra.tcm.sequences.LockedRanges;
+import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
+import org.apache.cassandra.tcm.serialization.Version;
+
+public class PrepareDropAccordTable implements Transformation
+{
+ public static final Serializer serializer = new Serializer();
+
+ public final TableReference tableRef;
+
+ public PrepareDropAccordTable(TableReference tableRef)
+ {
+ this.tableRef = tableRef;
+ }
+
+ @Override
+ public Kind kind()
+ {
+ return Kind.PREPARE_DROP_ACCORD_TABLE;
+ }
+
+ @Override
+ public Result execute(ClusterMetadata prev)
+ {
+ TableMetadata metadata = prev.schema.getKeyspaces().getTableOrViewNullable(tableRef.id);
+ if (metadata == null)
+ return new Rejected(ExceptionCode.INVALID, "Table " + tableRef + " is not known");
+ if (!metadata.isAccordEnabled())
+ return new Rejected(ExceptionCode.INVALID, "Table " + metadata + " is not an Accord table and should be dropped normally");
+ if (metadata.params.pendingDrop)
+ return new Rejected(ExceptionCode.INVALID, "Table " + metadata + " is in the process of being dropped");
+
+ KeyspaceMetadata ks = prev.schema.getKeyspaceMetadata(metadata.keyspace);
+ metadata = metadata.unbuild().params(metadata.params.unbuild().pendingDrop(true).build()).build();
+ ks = ks.withSwapped(ks.tables.withSwapped(metadata));
+
+ DropAccordTable operation = DropAccordTable.newSequence(tableRef, prev.nextEpoch());
+ ClusterMetadata.Transformer proposed = prev.transformer()
+ .with(new DistributedSchema(prev.schema.getKeyspaces().withAddedOrUpdated(ks)))
+ .with(prev.inProgressSequences.with(tableRef, operation));
+ return Transformation.success(proposed, LockedRanges.AffectedRanges.EMPTY);
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ PrepareDropAccordTable that = (PrepareDropAccordTable) o;
+ return tableRef.equals(that.tableRef);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(tableRef);
+ }
+
+ public static class Serializer implements AsymmetricMetadataSerializer<Transformation, PrepareDropAccordTable>
+ {
+ @Override
+ public void serialize(Transformation t, DataOutputPlus out, Version version) throws IOException
+ {
+ PrepareDropAccordTable plan = (PrepareDropAccordTable) t;
+ TableReference.serializer.serialize(plan.tableRef, out, version);
+ }
+
+ @Override
+ public PrepareDropAccordTable deserialize(DataInputPlus in, Version version) throws IOException
+ {
+ TableReference table = TableReference.serializer.deserialize(in, version);
+ return new PrepareDropAccordTable(table);
+ }
+
+ @Override
+ public long serializedSize(Transformation t, Version version)
+ {
+ PrepareDropAccordTable plan = (PrepareDropAccordTable) t;
+ return TableReference.serializer.serializedSize(plan.tableRef, version);
+ }
+ }
+}
diff --git a/src/java/org/apache/cassandra/tools/NodeTool.java b/src/java/org/apache/cassandra/tools/NodeTool.java
index f7e27ed231..874e3a1a50 100644
--- a/src/java/org/apache/cassandra/tools/NodeTool.java
+++ b/src/java/org/apache/cassandra/tools/NodeTool.java
@@ -262,7 +262,8 @@ public class NodeTool
.withCommand(CMSAdmin.InitializeCMS.class)
.withCommand(CMSAdmin.ReconfigureCMS.class)
.withCommand(CMSAdmin.Snapshot.class)
- .withCommand(CMSAdmin.Unregister.class);
+ .withCommand(CMSAdmin.Unregister.class)
+ .withCommand(CMSAdmin.ResumeDropAccordTable.class);
builder.withGroup("consensus_admin")
.withDescription("List and mark ranges as migrating between consensus protocols")
diff --git a/src/java/org/apache/cassandra/tools/nodetool/CMSAdmin.java b/src/java/org/apache/cassandra/tools/nodetool/CMSAdmin.java
index 0ed853ba68..bffee2654b 100644
--- a/src/java/org/apache/cassandra/tools/nodetool/CMSAdmin.java
+++ b/src/java/org/apache/cassandra/tools/nodetool/CMSAdmin.java
@@ -194,4 +194,16 @@ public abstract class CMSAdmin extends NodeTool.NodeToolCmd
probe.getCMSOperationsProxy().unregisterLeftNodes(nodeIds);
}
}
+
+ @Command(name = "resumedropaccordtable", description = "Resume a drop accord table operation which has stalled")
+ public static class ResumeDropAccordTable extends NodeTool.NodeToolCmd
+ {
+ @Arguments(usage = "[tableId]", description = "Table id of the table being dropped")
+ private String tableId;
+ @Override
+ public void execute(NodeProbe probe)
+ {
+ probe.getCMSOperationsProxy().resumeDropAccordTable(tableId);
+ }
+ }
}
diff --git a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
index 41d552bcf3..5af107a9dc 100644
--- a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
+++ b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
@@ -31,6 +31,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
@@ -52,6 +53,7 @@ import org.slf4j.LoggerFactory;
import accord.primitives.TxnId;
import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.ICluster;
@@ -74,6 +76,8 @@ import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.RequestCallback;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
@@ -1500,5 +1504,11 @@ public class ClusterUtils
}
}
}
+
+ public static TableId tableId(Cluster cluster, String ks, String table)
+ {
+ String str = cluster.getFirstRunningInstance().callOnInstance(() -> Schema.instance.getKeyspaceInstance(ks).getColumnFamilyStore(table).getTableId().toString());
+ return TableId.fromUUID(UUID.fromString(str));
+ }
}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDropKeyspaceTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDropKeyspaceTest.java
new file mode 100644
index 0000000000..353980f55f
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDropKeyspaceTest.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.accord;
+
+import java.io.IOException;
+
+import org.junit.Test;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.schema.TableId;
+
+public class AccordDropKeyspaceTest extends AccordDropTableBase
+{
+ @Test
+ public void dropKeyspace() throws IOException
+ {
+ int examples = 5;
+ int steps = 5;
+ try (Cluster cluster = Cluster.build(3)
+ .withoutVNodes()
+ .withConfig(c -> c.with(Feature.values())
+ .set("auto_snapshot", false))
+ .start())
+ {
+ fixDistributedSchemas(cluster);
+ for (int i = 0; i < examples; i++)
+ {
+ int j = 0;
+ try
+ {
+ addChaos(cluster, i);
+ init(cluster);
+ TableId id = createTable(cluster);
+ for (j = 0; j < steps; j++)
+ doTxn(cluster, j);
+ dropKeyspace(cluster);
+ validateAccord(cluster, id);
+ }
+ catch (Throwable t)
+ {
+ throw new AssertionError("Error at example " + i + ", " + j, t);
+ }
+ }
+ }
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDropTableBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDropTableBase.java
new file mode 100644
index 0000000000..732fe303cb
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDropTableBase.java
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.accord;
+
+import java.util.UUID;
+
+import com.google.common.base.Throwables;
+
+import accord.api.Key;
+import accord.local.CommandStores;
+import accord.local.KeyHistory;
+import accord.local.PreLoadContext;
+import accord.local.cfk.CommandsForKey;
+import accord.primitives.Ranges;
+import accord.primitives.TxnId;
+import accord.utils.async.AsyncChains;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.ICoordinator;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.shared.ClusterUtils;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.net.Verb;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.AccordCommandStore;
+import org.apache.cassandra.service.accord.AccordSafeCommandStore;
+import org.apache.cassandra.service.accord.AccordSafeCommandsForKey;
+import org.apache.cassandra.service.accord.AccordService;
+import org.apache.cassandra.service.accord.TokenRange;
+import org.assertj.core.api.Assertions;
+
+import static org.apache.cassandra.service.accord.AccordTestUtils.wrapInTxn;
+
+public class AccordDropTableBase extends TestBaseImpl
+{
+ protected static void addChaos(Cluster cluster, int example)
+ {
+ cluster.filters().reset();
+ cluster.filters().verbs(Verb.ACCORD_APPLY_REQ.id).from(1).to(3).drop();
+ }
+
+ protected static void doTxn(Cluster cluster, int step)
+ {
+ int stepId = step % 3;
+ int partitionId = step % 10;
+ int coordinatorId = (step % 2) + 1; // avoid node3 as it can't get applies from node1, so leads to user errors
+ ICoordinator coordinator = cluster.coordinator(coordinatorId);
+ switch (stepId)
+ {
+ case 0: // insert
+ retry(3, () -> coordinator.executeWithResult(wrapInTxn(withKeyspace("INSERT INTO %s.tbl(pk, v) VALUES (?, ?);")), ConsistencyLevel.ANY, partitionId, step));
+ break;
+ case 1: // insert + read
+ retry(3, () -> coordinator.executeWithResult(wrapInTxn(withKeyspace("UPDATE %s.tbl SET v+=1 WHERE pk=?;")), ConsistencyLevel.ANY, partitionId));
+ break;
+ case 2: // read
+ retry(3, () -> coordinator.executeWithResult(wrapInTxn(withKeyspace("SELECT * FROM %s.tbl WHERE pk=?")), ConsistencyLevel.ANY, partitionId));
+ break;
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ protected static void retry(int maxAttempts, Runnable fn)
+ {
+ for (int i = 0; i < maxAttempts; i++)
+ {
+ try
+ {
+ fn.run();
+ }
+ catch (Throwable t)
+ {
+ if (i == (maxAttempts - 1))
+ throw t;
+ }
+ }
+ }
+
+ protected static TableId createTable(Cluster cluster)
+ {
+ cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl(pk int PRIMARY KEY, v int) WITH transactional_mode='full'"));
+ return ClusterUtils.tableId(cluster, KEYSPACE, "tbl");
+ }
+
+ protected void dropKeyspace(Cluster cluster)
+ {
+ // drop keyspace should be rejected as there is an accord table... so validate that is true then do both
+ try
+ {
+ cluster.schemaChange(withKeyspace("DROP KEYSPACE %s"));
+ }
+ catch (Throwable t)
+ {
+ Assertions.assertThat(Throwables.getRootCause(t))
+ .hasMessage("Cannot drop keyspace 'distributed_test_keyspace' as it contains accord tables. (distributed_test_keyspace.tbl)");
+ }
+
+ // now do it for real
+ dropTable(cluster);
+ cluster.schemaChange(withKeyspace("DROP KEYSPACE %s"));
+ }
+
+ protected static void dropTable(Cluster cluster)
+ {
+ cluster.schemaChange(withKeyspace("DROP TABLE %s.tbl"));
+ }
+
+ protected static void validateAccord(Cluster cluster, TableId id)
+ {
+ String s = id.toString();
+ for (IInvokableInstance inst : cluster)
+ {
+ inst.runOnInstance(() -> {
+ TableId tableId = TableId.fromUUID(UUID.fromString(s));
+ AccordService accord = (AccordService) AccordService.instance();
+ PreLoadContext ctx = PreLoadContext.contextFor(Ranges.single(TokenRange.fullRange(tableId)), KeyHistory.COMMANDS);
+ CommandStores stores = accord.node().commandStores();
+ for (int storeId : stores.ids())
+ {
+ AccordCommandStore store = (AccordCommandStore) stores.forId(storeId);
+ AsyncChains.getUnchecked(store.submit(ctx, input -> {
+ AccordSafeCommandStore safe = (AccordSafeCommandStore) input;
+ for (Key key : safe.commandsForKeysKeys())
+ {
+ AccordSafeCommandsForKey safeCFK = safe.maybeCommandsForKey(key);
+ if (safeCFK == null) // we read and found a key, but its null at load time... so ignore it
+ continue;
+ CommandsForKey cfk = safeCFK.current();
+ CommandsForKey.TxnInfo minUndecided = cfk.minUndecided();
+ if (minUndecided != null)
+ throw new AssertionError("Undecided txn: " + minUndecided);
+ TxnId next = cfk.nextWaitingToApply();
+ if (next != null)
+ throw new AssertionError("Unapplied txn: " + next);
+ }
+ return null;
+ }));
+ }
+ });
+ }
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDropTableTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDropTableTest.java
new file mode 100644
index 0000000000..b13bcb05ce
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDropTableTest.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.accord;
+
+import java.io.IOException;
+
+import org.junit.Test;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.schema.TableId;
+
+public class AccordDropTableTest extends AccordDropTableBase
+{
+ @Test
+ public void dropTable() throws IOException
+ {
+ int examples = 5;
+ int steps = 5;
+ try (Cluster cluster = Cluster.build(3)
+ .withoutVNodes()
+ .withConfig(c -> c.with(Feature.values())
+ .set("auto_snapshot", false))
+ .start())
+ {
+ fixDistributedSchemas(cluster);
+ init(cluster);
+ for (int i = 0; i < examples; i++)
+ {
+ int j = 0;
+ try
+ {
+ addChaos(cluster, i);
+ TableId id = createTable(cluster);
+ for (j = 0; j < steps; j++)
+ doTxn(cluster, j);
+ dropTable(cluster);
+ validateAccord(cluster, id);
+ }
+ catch (Throwable t)
+ {
+ throw new AssertionError("Error at example " + i + ", " + j, t);
+ }
+ }
+ }
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java b/test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java
index cc29ba27f8..81170ebd79 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java
@@ -59,6 +59,7 @@ import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.SchemaTransformation;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.accord.AccordStaleReplicas;
+import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState;
import org.apache.cassandra.tcm.AtomicLongBackedProcessor;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
@@ -157,41 +158,22 @@ public class ClusterMetadataTestHelper
AccordFastPath.EMPTY,
LockedRanges.EMPTY,
InProgressSequences.EMPTY,
- null,
+ ConsensusMigrationState.EMPTY,
ImmutableMap.of(),
AccordStaleReplicas.EMPTY);
}
public static ClusterMetadata minimalForTesting(IPartitioner partitioner)
{
- return new ClusterMetadata(Epoch.EMPTY,
- partitioner,
- null,
- null,
- null,
- DataPlacements.empty(),
- AccordFastPath.EMPTY,
- null,
- null,
- null,
- ImmutableMap.of(),
- AccordStaleReplicas.EMPTY);
+ return minimalForTesting(Epoch.EMPTY, partitioner);
}
public static ClusterMetadata minimalForTesting(Keyspaces keyspaces)
{
- return new ClusterMetadata(Epoch.EMPTY,
- Murmur3Partitioner.instance,
- new DistributedSchema(keyspaces),
- null,
- null,
- DataPlacements.empty(),
- AccordFastPath.EMPTY,
- null,
- null,
- null,
- ImmutableMap.of(),
- AccordStaleReplicas.EMPTY);
+ return minimalForTesting(Murmur3Partitioner.instance).transformer()
+ .with(new DistributedSchema(keyspaces))
+ .build()
+ .metadata.forceEpoch(Epoch.EMPTY);
}
public static ClusterMetadataService syncInstanceForTest()
diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/AccordTopologyMixupTest.java b/test/distributed/org/apache/cassandra/fuzz/topology/AccordTopologyMixupTest.java
index 9a222c7c05..991dbfd346 100644
--- a/test/distributed/org/apache/cassandra/fuzz/topology/AccordTopologyMixupTest.java
+++ b/test/distributed/org/apache/cassandra/fuzz/topology/AccordTopologyMixupTest.java
@@ -23,6 +23,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutionException;
import java.util.function.BiFunction;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -56,8 +57,10 @@ import org.apache.cassandra.distributed.api.SimpleQueryResult;
import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.distributed.test.accord.AccordTestBase;
import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.api.AccordAgent;
import org.apache.cassandra.service.consensus.TransactionalMode;
+import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.utils.ASTGenerators;
import org.apache.cassandra.utils.CassandraGenerators;
import org.apache.cassandra.utils.Isolated;
@@ -219,6 +222,24 @@ public class AccordTopologyMixupTest extends TopologyMixupTestBase<AccordTopolog
.set("paxos_variant", Config.PaxosVariant.v2.name());
}
+ @Override
+ protected void onStartupComplete(long tcmEpoch)
+ {
+ cluster.forEach(i -> {
+ if (i.isShutdown()) return;
+ i.runOnInstance(() -> {
+ try
+ {
+ AccordService.instance().epochReady(Epoch.create(tcmEpoch)).get();
+ }
+ catch (InterruptedException | ExecutionException e)
+ {
+ throw new RuntimeException(e);
+ }
+ });
+ });
+ }
+
@Override
public void debugTxn(@Nullable Node.Id exclude, String type, TxnId txnId)
{
diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/TopologyMixupTestBase.java b/test/distributed/org/apache/cassandra/fuzz/topology/TopologyMixupTestBase.java
index f3ef929eec..fd669340ae 100644
--- a/test/distributed/org/apache/cassandra/fuzz/topology/TopologyMixupTestBase.java
+++ b/test/distributed/org/apache/cassandra/fuzz/topology/TopologyMixupTestBase.java
@@ -69,15 +69,21 @@ import org.apache.cassandra.distributed.impl.InstanceConfig;
import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.metrics.TCMMetrics;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
+import org.apache.cassandra.tcm.Epoch;
+import org.apache.cassandra.tcm.Retry;
+import org.apache.cassandra.tcm.log.Entry;
+import org.apache.cassandra.tcm.log.LogState;
import org.apache.cassandra.utils.ConfigGenBuilder;
import static accord.utils.Property.commands;
import static accord.utils.Property.ignoreCommand;
import static accord.utils.Property.multistep;
import static accord.utils.Property.stateful;
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
/**
* These tests can create many instances, so mac users may need to run the following to avoid address bind failures
@@ -482,12 +488,24 @@ public abstract class TopologyMixupTestBase<S extends TopologyMixupTestBase.Sche
sb.append("Yaml Config:\n").append(YamlConfigurationLoader.toYaml(this.yamlConfigOverrides));
sb.append("\nTopology:\n").append(topologyHistory);
sb.append("\nCMS Voting Group: ").append(Arrays.toString(cmsGroup));
+ if (epochHistory != null)
+ sb.append("\n").append(epochHistory);
return sb.toString();
}
+ private String epochHistory = null;
+
@Override
public void close() throws Exception
{
+ epochHistory = cluster.get(cmsGroup[0]).callOnInstance(() -> {
+ LogState all = ClusterMetadataService.instance().processor().reconstruct(Epoch.EMPTY, Epoch.create(Long.MAX_VALUE), Retry.Deadline.retryIndefinitely(DatabaseDescriptor.getCmsAwaitTimeout().to(NANOSECONDS),
+ TCMMetrics.instance.commitRetries));
+ StringBuilder sb = new StringBuilder("Epochs:");
+ for (Entry e : all.entries)
+ sb.append("\n").append(e.epoch.getEpoch()).append(": ").append(e.transform);
+ return sb.toString();
+ });
cluster.close();
}
}
diff --git a/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java b/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
index 70e49bc7fe..c6559ee851 100644
--- a/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
+++ b/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
@@ -218,6 +218,7 @@ public class PreparedStatementsTest extends CQLTester
session.execute(preparedBatch.bind(2, 2, "value2"));
session.execute(preparedTxn.bind(3, 3, "value3"));
+ sessionSchemaUpdate(session, dropTableStatement); // since this is an accord table, need to drop the table before the keyspace
sessionSchemaUpdate(session, dropKsStatement);
sessionSchemaUpdate(session, createKsStatement);
sessionSchemaUpdate(session, createTableStatement);
@@ -229,6 +230,7 @@ public class PreparedStatementsTest extends CQLTester
session.execute(prepared.bind(1, 1, "value"));
session.execute(preparedBatch.bind(2, 2, "value2"));
session.execute(preparedTxn.bind(3, 3, "value3"));
+ sessionSchemaUpdate(session, dropTableStatement); // since this is an accord table, need to drop the table before the keyspace
sessionSchemaUpdate(session, dropKsStatement);
}
@@ -249,6 +251,7 @@ public class PreparedStatementsTest extends CQLTester
Session session = sessionNet(version);
String createTableStatement = "CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".qp_cleanup (a int PRIMARY KEY, b int, c int) WITH transactional_mode='unsafe';";
String alterTableStatement = "ALTER TABLE " + KEYSPACE + ".qp_cleanup ADD d int;";
+ String dropTableStatement = "DROP TABLE IF EXISTS " + KEYSPACE + ".qp_cleanup;";
sessionSchemaUpdate(session, dropKsStatement);
sessionSchemaUpdate(session, createKsStatement);
@@ -315,6 +318,7 @@ public class PreparedStatementsTest extends CQLTester
}
}
+ sessionSchemaUpdate(session, dropTableStatement);
sessionSchemaUpdate(session, dropKsStatement);
}
@@ -335,6 +339,7 @@ public class PreparedStatementsTest extends CQLTester
Session session = sessionNet(version);
String createTableStatement = "CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".qp_cleanup (a int PRIMARY KEY, b int, c int) WITH transactional_mode='unsafe';";
String alterTableStatement = "ALTER TABLE " + KEYSPACE + ".qp_cleanup ADD d int;";
+ String dropTableStatement = "DROP TABLE IF EXISTS " + KEYSPACE + ".qp_cleanup;";
sessionSchemaUpdate(session, dropKsStatement);
sessionSchemaUpdate(session, createKsStatement);
@@ -382,6 +387,7 @@ public class PreparedStatementsTest extends CQLTester
Assertions.assertThat(columnNames(rs)).containsExactlyInAnyOrder("a", "b", "c");
}
+ sessionSchemaUpdate(session, dropTableStatement);
sessionSchemaUpdate(session, dropKsStatement);
}
diff --git a/test/unit/org/apache/cassandra/schema/TableParamsTest.java b/test/unit/org/apache/cassandra/schema/TableParamsTest.java
new file mode 100644
index 0000000000..e8bf30fe0a
--- /dev/null
+++ b/test/unit/org/apache/cassandra/schema/TableParamsTest.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.schema;
+
+import org.junit.Test;
+
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.tcm.membership.NodeVersion;
+import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializers;
+import org.apache.cassandra.utils.CassandraGenerators.TableParamsBuilder;
+import org.apache.cassandra.utils.FailingConsumer;
+import org.quicktheories.core.Gen;
+
+import static org.quicktheories.QuickTheory.qt;
+
+
+public class TableParamsTest
+{
+ @Test
+ public void serdeLatest()
+ {
+ DataOutputBuffer output = new DataOutputBuffer();
+ qt().forAll(tableParams()).checkAssert(FailingConsumer.orFail(params -> {
+ AsymmetricMetadataSerializers.testSerde(output, TableParams.serializer, params, NodeVersion.CURRENT_METADATA_VERSION);
+ }));
+ }
+
+ private static Gen<TableParams> tableParams()
+ {
+ return new TableParamsBuilder()
+ .withKnownMemtables()
+ .withTransactionalMode()
+ .withFastPathStrategy()
+ .build();
+ }
+}
\ No newline at end of file
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java b/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java
index 02d823cfd4..3939371ad9 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java
@@ -135,7 +135,7 @@ public class AccordKeyspaceTest extends CQLTester.InMemory
public void findOverlappingKeys()
{
var tableIdGen = fromQT(CassandraGenerators.TABLE_ID_GEN);
- var partitionGen = fromQT(CassandraGenerators.partitioners());
+ var partitionGen = fromQT(CassandraGenerators.partitioners()).map(CassandraGenerators::simplify);
var sstableFormats = DatabaseDescriptor.getSSTableFormats();
List<String> sstableFormatNames = new ArrayList<>(sstableFormats.keySet());
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordStaleReplicasTest.java b/test/unit/org/apache/cassandra/service/accord/AccordStaleReplicasTest.java
index 83eced9b66..bce9a3fb14 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordStaleReplicasTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordStaleReplicasTest.java
@@ -47,7 +47,7 @@ public class AccordStaleReplicasTest
qt().check(rs -> {
Epoch epoch = epochGen.next(rs);
Set<Node.Id> nodes = nodesGen.next(rs);
- AsymmetricMetadataSerializers.testSerde(buffer, AccordStaleReplicas.serializer, new AccordStaleReplicas(nodes, epoch), Version.V2);
+ AsymmetricMetadataSerializers.testSerde(buffer, AccordStaleReplicas.serializer, new AccordStaleReplicas(nodes, epoch), Version.MIN_ACCORD_VERSION);
});
}
}
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java b/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java
index 12d5c75c01..5969770e2e 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java
@@ -58,7 +58,6 @@ import accord.utils.RandomSource;
import org.apache.cassandra.concurrent.AdaptingScheduledExecutorPlus;
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.gms.EndpointState;
import org.apache.cassandra.gms.Gossiper;
@@ -70,8 +69,8 @@ import org.apache.cassandra.net.ConnectionType;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessageDelivery;
import org.apache.cassandra.net.RequestCallback;
-import org.apache.cassandra.tcm.ClusterMetadataService;
-import org.apache.cassandra.tcm.StubClusterMetadataService;
+import org.apache.cassandra.tcm.ValidatingClusterMetadataService;
+import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.utils.AccordGenerators;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.concurrent.Future;
@@ -86,9 +85,7 @@ public class AccordSyncPropagatorTest
public static void setup() throws NoSuchFieldException, IllegalAccessException
{
DatabaseDescriptor.daemonInitialization();
- DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
- ClusterMetadataService.unsetInstance();
- ClusterMetadataService.setInstance(StubClusterMetadataService.forTesting());
+ ValidatingClusterMetadataService.createAndRegister(Version.MIN_ACCORD_VERSION);
}
@Test
diff --git a/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java b/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java
index fc2a36d8ab..f5faef62c9 100644
--- a/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java
@@ -49,8 +49,8 @@ import org.slf4j.LoggerFactory;
import accord.api.ConfigurationService;
import accord.api.ConfigurationService.EpochReady;
-import accord.api.Scheduler;
import accord.api.LocalConfig;
+import accord.api.Scheduler;
import accord.impl.SizeOfIntersectionSorter;
import accord.impl.TestAgent;
import accord.local.Node;
@@ -79,11 +79,9 @@ import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.SimulatedMessageDelivery;
import org.apache.cassandra.net.SimulatedMessageDelivery.Action;
import org.apache.cassandra.net.Verb;
-import org.apache.cassandra.schema.DistributedMetadataLogKeyspace;
import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.KeyspaceParams;
-import org.apache.cassandra.schema.Keyspaces;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableParams;
@@ -96,11 +94,12 @@ import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.MultiStepOperation;
import org.apache.cassandra.tcm.StubClusterMetadataService;
import org.apache.cassandra.tcm.Transformation;
-import org.apache.cassandra.tcm.membership.Directory;
+import org.apache.cassandra.tcm.ValidatingClusterMetadataService;
import org.apache.cassandra.tcm.membership.Location;
import org.apache.cassandra.tcm.membership.NodeAddresses;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.membership.NodeState;
+import org.apache.cassandra.tcm.membership.NodeVersion;
import org.apache.cassandra.tcm.ownership.UniformRangePlacement;
import org.apache.cassandra.tcm.sequences.LeaveStreams;
import org.apache.cassandra.tcm.transformations.PrepareJoin;
@@ -135,13 +134,13 @@ public class EpochSyncTest
})
.addAllIf(Cluster::hasPendingWork, b ->
b.addIf(c -> !c.status(s -> s == Cluster.Status.Registered).isEmpty(), (rs, state) -> {
- long epoch = state.current.epoch.getEpoch() + 1;
+ long epoch = state.cms.metadata().epoch.getEpoch() + 1;
Node.Id pick = rs.pick(state.status(s -> s == Cluster.Status.Registered));
return new SimpleCommand<>(pick + " Start Joining; epoch=" + epoch,
c -> c.increment(pick));
})
- .addIf(c -> !c.current.inProgressSequences.isEmpty(),
- (rs, state) -> new SimpleCommand<>("Next Epoch Step; epoch=" + (state.current.epoch.getEpoch() + 1),
+ .addIf(c -> !c.cms.metadata().inProgressSequences.isEmpty(),
+ (rs, state) -> new SimpleCommand<>("Next Epoch Step; epoch=" + (state.cms.metadata().epoch.getEpoch() + 1),
Cluster::incrementInProgressSequences))
)
.addAllIf(Cluster::hasNoPendingWork, b ->
@@ -150,7 +149,7 @@ public class EpochSyncTest
)
.addIf(Cluster::hasWork, EpochSyncTest::processSome)
.add(rs -> new SimpleCommand<>("Validate", c -> c.validate(false)))
- .add((rs, cluster) -> new SimpleCommand<>("Bump Epoch " + (cluster.current.epoch.getEpoch() + 1), Cluster::bumpEpoch))
+ .add((rs, cluster) -> new SimpleCommand<>("Bump Epoch " + (cluster.cms.metadata().epoch.getEpoch() + 1), Cluster::bumpEpoch))
.build());
}
@@ -159,7 +158,7 @@ public class EpochSyncTest
List<Node.Id> registered = cluster.status(s -> s == Cluster.Status.Registered);
if (!registered.isEmpty())
registered.forEach(cluster::increment);
- while (!cluster.current.inProgressSequences.isEmpty())
+ while (!cluster.cms.metadata().inProgressSequences.isEmpty())
cluster.incrementInProgressSequences();
}
@@ -169,7 +168,7 @@ public class EpochSyncTest
long token = cluster.tokenGen.nextLong(rs);
while (cluster.tokens.contains(token))
token = cluster.tokenGen.nextLong(rs);
- long epoch = cluster.current.epoch.getEpoch() + 1;
+ long epoch = cluster.cms.metadata().epoch.getEpoch() + 1;
long finalToken = token;
return new SimpleCommand<>("Start Node " + id + "; token=" + token + ", epoch=" + epoch,
c -> c.registerNode(id, finalToken));
@@ -180,7 +179,7 @@ public class EpochSyncTest
List<Node.Id> alive = cluster.joined();
Node.Id pick = rs.pick(alive);
long token = cluster.instances.get(pick).token;
- long epoch = cluster.current.epoch.getEpoch() + 1;
+ long epoch = cluster.cms.metadata().epoch.getEpoch() + 1;
return new SimpleCommand<>("Remove Node " + pick + "; token=" + token + "; epoch=" + epoch, c -> c.removeNode(pick));
}
@@ -207,10 +206,7 @@ public class EpochSyncTest
private final SimulatedExecutorFactory globalExecutor;
private final ScheduledExecutorPlus scheduler;
private int nodeCounter = 0;
- private ClusterMetadata current = new ClusterMetadata(Murmur3Partitioner.instance, Directory.EMPTY,
- new DistributedSchema(Keyspaces.of(
- DistributedMetadataLogKeyspace.initialMetadata(Collections.singleton("dc1")),
- KeyspaceMetadata.create("test", KeyspaceParams.simple(rf), Tables.of(TableMetadata.minimal("test", "tb1").unbuild().params(TableParams.builder().transactionalMode(TransactionalMode.full).build()).build())))));
+ private final ValidatingClusterMetadataService cms = ValidatingClusterMetadataService.createAndRegister(NodeVersion.CURRENT_METADATA_VERSION);
private final IFailureDetector fd = new IFailureDetector()
{
@Override
@@ -258,6 +254,8 @@ public class EpochSyncTest
public Cluster(RandomSource rs)
{
+ // add the test keyspace
+ createTestKeyspaceAndTable();
this.rs = rs;
this.minNodes = 3;
this.maxNodes = 10;
@@ -307,7 +305,7 @@ public class EpochSyncTest
private boolean hasPendingWork()
{
return !status(s -> s == Cluster.Status.Registered).isEmpty()
- || !current.inProgressSequences.isEmpty();
+ || !cms.metadata().inProgressSequences.isEmpty();
}
private boolean hasNoPendingWork()
@@ -317,7 +315,7 @@ public class EpochSyncTest
private Transformation.Success process(Transformation transformation)
{
- Transformation.Result result = transformation.execute(current);
+ Transformation.Result result = transformation.execute(cms.metadata());
if (result.isRejected())
throw new IllegalStateException("Unable to make TCM transition: " + result.rejected());
return result.success();
@@ -325,7 +323,7 @@ public class EpochSyncTest
private Transformation.Success process(MultiStepOperation<?> transformation)
{
- Transformation.Result result = transformation.applyTo(current);
+ Transformation.Result result = transformation.applyTo(cms.metadata());
if (result.isRejected())
throw new IllegalStateException("Unable to make TCM transition");
return result.success();
@@ -333,11 +331,11 @@ public class EpochSyncTest
public void incrementInProgressSequences()
{
- if (current.inProgressSequences.isEmpty())
+ if (cms.metadata().inProgressSequences.isEmpty())
throw new IllegalStateException("Attempted to bump epoch when nothing was pending");
- Iterator<MultiStepOperation<?>> it = current.inProgressSequences.iterator();
+ Iterator<MultiStepOperation<?>> it = cms.metadata().inProgressSequences.iterator();
Invariants.checkState(it.hasNext());
- notify(current = process(it.next()).metadata);
+ notify(process(it.next()).metadata);
}
private static boolean left(ClusterMetadata metadata, Node.Id id)
@@ -372,6 +370,21 @@ public class EpochSyncTest
return new Node.Id(ByteArrayUtil.getInt(address.addressBytes));
}
+ private void createTestKeyspaceAndTable()
+ {
+ ClusterMetadata current = cms.metadata();
+ Tables tables = Tables.of(TableMetadata.minimal("test", "tb1").unbuild()
+ .partitioner(Murmur3Partitioner.instance)
+ .params(TableParams.builder().transactionalMode(TransactionalMode.full).build())
+ .build());
+ KeyspaceMetadata ks = KeyspaceMetadata.create("test", KeyspaceParams.simple(rf), tables);
+
+ cms.setMetadata(current.transformer()
+ .with(new DistributedSchema(current.schema.getKeyspaces().with(ks)))
+ .build()
+ .metadata);
+ }
+
void validate(boolean isDone)
{
for (Node.Id id : notRemoved())
@@ -380,7 +393,7 @@ public class EpochSyncTest
if (removed.contains(id)) continue; // ignore removed nodes
AccordConfigurationService conf = inst.config;
TopologyManager tm = inst.topology;
- for (long epoch = inst.epoch.getEpoch(); epoch <= current.epoch.getEpoch(); epoch++)
+ for (long epoch = inst.epoch.getEpoch(); epoch <= cms.metadata().epoch.getEpoch(); epoch++)
{
// validate config
EpochSnapshot snapshot = conf.getEpochSnapshot(epoch);
@@ -550,15 +563,14 @@ public class EpochSyncTest
Invariants.checkState(!tokens.contains(token), "Attempted to add token %d for node %s but token is already taken", token, id);
Invariants.checkState(!instances.containsKey(id), "Attempted to add node %s; but already exists", id);
- Epoch epoch = Epoch.create(current.epoch.getEpoch() + 1);
+ ClusterMetadata.Transformer builder = cms.metadata().transformer();
- Instance instance = new Instance(id, token, epoch, createMessaging(id), fd);
+ Instance instance = new Instance(id, token, builder.epoch(), createMessaging(id), fd);
instances.put(id, instance);
tokens.add(token);
- current = current.forceEpoch(epoch)
- .withDirectory(current.directory.with(new NodeAddresses(address(id)), new Location("dc1", "r1")));
- notify(current);
+ builder.register(new NodeAddresses(address(id)), new Location("dc1", "r1"), NodeVersion.CURRENT);
+ notify(builder.build().metadata);
}
void increment(Node.Id pick)
@@ -574,7 +586,7 @@ public class EpochSyncTest
case Registered:
inst.status = Status.Joining;
PrepareJoin task = new PrepareJoin(new NodeId(pick.id), Collections.singleton(new LongToken(inst.token)), new UniformRangePlacement(), true, false);
- notify(current = process(task).metadata);
+ notify(process(task).metadata);
break;
default:
throw new UnsupportedOperationException("Unknown status: " + inst.status);
@@ -588,13 +600,12 @@ public class EpochSyncTest
removed.add(pick);
inst.status = Status.Leaving;
PrepareLeave prepareLeave = new PrepareLeave(new NodeId(pick.id), false, new UniformRangePlacement(), LeaveStreams.Kind.REMOVENODE);
- notify(current = process(prepareLeave).metadata);
+ notify(process(prepareLeave).metadata);
}
void bumpEpoch()
{
- current = current.forceEpoch(Epoch.create(current.epoch.getEpoch() + 1));
- notify(current);
+ notify(cms.metadata().forceEpoch(Epoch.create(cms.metadata().epoch.getEpoch() + 1)));
}
private void notify(ClusterMetadata current)
@@ -603,7 +614,7 @@ public class EpochSyncTest
Ranges ranges = t.ranges().mergeTouching();
if (!current.placements.get(replication_params).reads.isEmpty())
Assertions.assertThat(ranges).hasSize(1);
- ((StubClusterMetadataService) ClusterMetadataService.instance()).setMetadata(current);
+ cms.setMetadata(current);
for (Node.Id id : status(s -> s != Status.Removed))
{
Instance inst = instances.get(id);
diff --git a/test/unit/org/apache/cassandra/service/accord/FetchMinEpochTest.java b/test/unit/org/apache/cassandra/service/accord/FetchMinEpochTest.java
index 936bc13b0b..53319a7ec2 100644
--- a/test/unit/org/apache/cassandra/service/accord/FetchMinEpochTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/FetchMinEpochTest.java
@@ -78,6 +78,7 @@ public class FetchMinEpochTest
{
DataOutputBuffer output = new DataOutputBuffer();
Gen<FetchMinEpoch> gen = fromQT(CassandraGenerators.partitioners())
+ .map(CassandraGenerators::simplify)
.flatMap(partitioner ->
Gens.lists(AccordGenerators.range(partitioner)
.map(r -> (TokenRange) r))
diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/AccordRoutingKeyByteSourceTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/AccordRoutingKeyByteSourceTest.java
index aaa1db454d..78dcb5b70c 100644
--- a/test/unit/org/apache/cassandra/service/accord/serializers/AccordRoutingKeyByteSourceTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/serializers/AccordRoutingKeyByteSourceTest.java
@@ -49,7 +49,7 @@ public class AccordRoutingKeyByteSourceTest
@Test
public void tokenSerde()
{
- qt().forAll(fromQT(token())).check(token -> {
+ qt().forAll(fromQT(CassandraGenerators.partitioners().map(CassandraGenerators::simplify).flatMap(CassandraGenerators::token))).check(token -> {
var serializer = AccordRoutingKeyByteSource.create(token.getPartitioner());
byte[] min = ByteSourceInverse.readBytes(serializer.minAsComparableBytes());
byte[] max = ByteSourceInverse.readBytes(serializer.maxAsComparableBytes());
diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/DepsSerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/DepsSerializerTest.java
index 4238f8a687..5283a8b6ea 100644
--- a/test/unit/org/apache/cassandra/service/accord/serializers/DepsSerializerTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/serializers/DepsSerializerTest.java
@@ -24,18 +24,15 @@ import org.junit.Test;
import accord.primitives.Deps;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.db.marshal.AbstractType;
-import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.dht.IPartitioner;
-import org.apache.cassandra.dht.LocalPartitioner;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.io.IVersionedSerializers;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaProvider;
-import org.apache.cassandra.utils.AbstractTypeGenerators;
import org.apache.cassandra.utils.AccordGenerators;
+import org.apache.cassandra.utils.CassandraGenerators;
import org.mockito.Mockito;
import static accord.utils.Property.qt;
@@ -56,7 +53,7 @@ public class DepsSerializerTest
{
DataOutputBuffer buffer = new DataOutputBuffer();
qt().check(rs -> {
- IPartitioner partitioner = AccordGenerators.partitioner().map(DepsSerializerTest::normalize).next(rs);
+ IPartitioner partitioner = AccordGenerators.partitioner().map(CassandraGenerators::simplify).next(rs);
Schema.instance = Mockito.mock(SchemaProvider.class);
DatabaseDescriptor.setPartitionerUnsafe(partitioner);
Mockito.when(Schema.instance.getExistingTablePartitioner(Mockito.any())).thenReturn(partitioner);
@@ -65,17 +62,4 @@ public class DepsSerializerTest
IVersionedSerializers.testSerde(buffer, DepsSerializer.deps, deps, version.value);
});
}
-
- private static IPartitioner normalize(IPartitioner partitioner)
- {
- // serializers require tokens to fit within 1 << 16, but that makes the test flakey when LocalPartitioner with a nested type is found...
- if (!(partitioner instanceof LocalPartitioner)) return partitioner;
- if (!shouldSimplify(partitioner.getTokenValidator())) return partitioner;
- return new LocalPartitioner(Int32Type.instance);
- }
-
- private static boolean shouldSimplify(AbstractType<?> type)
- {
- return AbstractTypeGenerators.contains(type, t -> t.isCollection());
- }
}
\ No newline at end of file
diff --git a/test/unit/org/apache/cassandra/tcm/ClusterMetadataMetadataKeyTest.java b/test/unit/org/apache/cassandra/tcm/ClusterMetadataMetadataKeyTest.java
new file mode 100644
index 0000000000..e475b6c850
--- /dev/null
+++ b/test/unit/org/apache/cassandra/tcm/ClusterMetadataMetadataKeyTest.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tcm;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.utils.FBUtilities;
+import org.assertj.core.api.Assertions;
+
+/**
+ * This test is to make sure that the fields of {@link ClusterMetadata} have a matching {@link MetadataKey} and the
+ * utility functions linking key to field are maintained.
+ *
+ * If this test is failing it likely means a new field was added to {@link ClusterMetadata} and {@link MetadataKeys} was
+ * not updated to know about it.
+ */
+public class ClusterMetadataMetadataKeyTest
+{
+ static
+ {
+ DatabaseDescriptor.clientInitialization();
+ }
+
+ private static final Map<String, Field> NAME_TO_KEY;
+
+ static
+ {
+ ImmutableMap.Builder<String, Field> builder = ImmutableMap.builder();
+ for (Field field : MetadataKeys.class.getDeclaredFields())
+ {
+ if (field.getType() == MetadataKey.class
+ && Modifier.isStatic(field.getModifiers())
+ && Modifier.isPublic(field.getModifiers()))
+ builder.put(field.getName(), field);
+ }
+ NAME_TO_KEY = builder.build();
+ }
+
+ @Test
+ public void metadataKeyExists() throws IllegalAccessException
+ {
+ ClusterMetadata empty = new ClusterMetadata(Murmur3Partitioner.instance);
+ // Theese are fields that should not have MetadataKeys and should be ignored.
+ Set<String> exclude = ImmutableSet.of("metadataIdentifier",
+ "epoch",
+ "partitioner",
+ "extensions");
+ // Mapping of ClusterMetadata field names to MetadataKey name; mapping is only needed if the names don't match.
+ Map<String, String> mapping = ImmutableMap.of("directory", "node_directory",
+ "placements", "data_placements");
+ for (Field field : ClusterMetadata.class.getDeclaredFields())
+ {
+ if (Modifier.isStatic(field.getModifiers())
+ || !Modifier.isPublic(field.getModifiers())
+ || !Modifier.isFinal(field.getModifiers()))
+ continue;
+ String name = field.getName();
+ if (exclude.contains(name)) continue;
+ if (mapping.containsKey(name))
+ name = mapping.get(name);
+ String snakeName = FBUtilities.camelToSnake(name).toUpperCase(Locale.ROOT);
+ Assertions.assertThat(NAME_TO_KEY.keySet())
+ .describedAs("Unable to locate MetadataKey for %s", snakeName)
+ .contains(snakeName);
+ MetadataKey expectedKey = (MetadataKey) NAME_TO_KEY.get(snakeName).get(null);
+ if (!MetadataKeys.CORE_METADATA.containsKey(expectedKey))
+ throw new IllegalStateException("MetadataKeys.CORE_METADATA is missing key " + expectedKey + " for field " + name);
+
+ Assertions.assertThat(field.get(empty))
+ .describedAs("Extraction function does not seem to match the field %s and key %s", name, snakeName)
+ .isSameAs(MetadataKeys.CORE_METADATA.get(expectedKey).apply(empty));
+ }
+ }
+}
diff --git a/test/unit/org/apache/cassandra/tcm/ClusterMetadataSerializerTest.java b/test/unit/org/apache/cassandra/tcm/ClusterMetadataSerializerTest.java
new file mode 100644
index 0000000000..1117b4ab88
--- /dev/null
+++ b/test/unit/org/apache/cassandra/tcm/ClusterMetadataSerializerTest.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tcm;
+
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.io.util.DataInputBuffer;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.service.accord.AccordFastPath;
+import org.apache.cassandra.service.accord.AccordStaleReplicas;
+import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState;
+import org.apache.cassandra.tcm.membership.NodeVersion;
+import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializers;
+import org.apache.cassandra.tcm.serialization.Version;
+import org.apache.cassandra.utils.CassandraGenerators.ClusterMetadataBuilder;
+import org.assertj.core.api.Assertions;
+import org.quicktheories.core.Gen;
+
+import static org.apache.cassandra.utils.FailingConsumer.orFail;
+import static org.quicktheories.QuickTheory.qt;
+
+public class ClusterMetadataSerializerTest
+{
+ static
+ {
+ DatabaseDescriptor.clientInitialization();
+ }
+
+ @Test
+ public void serdeLatest()
+ {
+ DataOutputBuffer output = new DataOutputBuffer();
+ qt().forAll(new ClusterMetadataBuilder().build()).checkAssert(orFail(cm -> {
+ AsymmetricMetadataSerializers.testSerde(output, ClusterMetadata.serializer, cm, NodeVersion.CURRENT_METADATA_VERSION);
+ }));
+ }
+
+ @Test
+ public void serdeWithoutAccord()
+ {
+ DataOutputBuffer output = new DataOutputBuffer();
+ Gen<ClusterMetadata> gen = new ClusterMetadataBuilder().build().assuming(cm -> {
+ if (!cm.consensusMigrationState.equals(ConsensusMigrationState.EMPTY))
+ return true;
+ if (!cm.accordStaleReplicas.equals(AccordStaleReplicas.EMPTY))
+ return true;
+ if (!cm.accordFastPath.equals(AccordFastPath.EMPTY))
+ return true;
+ return false;
+ });
+ qt().forAll(gen).checkAssert(orFail(cm -> {
+ output.clear();
+ Version version = Version.V2; // this is the version before accord
+ long expectedSize = ClusterMetadata.serializer.serializedSize(cm, version);
+ ClusterMetadata.serializer.serialize(cm, output, version);
+ Assertions.assertThat(output.getLength()).describedAs("The serialized size and bytes written do not match").isEqualTo(expectedSize);
+ DataInputBuffer in = new DataInputBuffer(output.unsafeGetBufferAndFlip(), false);
+ ClusterMetadata read = ClusterMetadata.serializer.deserialize(in, version);
+ Assertions.assertThat(read).isNotEqualTo(cm);
+
+ Assertions.assertThat(read.consensusMigrationState).isEqualTo(ConsensusMigrationState.EMPTY);
+ Assertions.assertThat(read.accordStaleReplicas).isEqualTo(AccordStaleReplicas.EMPTY);
+ Assertions.assertThat(read.accordFastPath).isEqualTo(AccordFastPath.EMPTY);
+ }));
+ }
+}
diff --git a/test/unit/org/apache/cassandra/tcm/ClusterMetadataTransformationTest.java b/test/unit/org/apache/cassandra/tcm/ClusterMetadataTransformationTest.java
index 109fefeda2..27b679d7f5 100644
--- a/test/unit/org/apache/cassandra/tcm/ClusterMetadataTransformationTest.java
+++ b/test/unit/org/apache/cassandra/tcm/ClusterMetadataTransformationTest.java
@@ -275,7 +275,7 @@ public class ClusterMetadataTransformationTest
// anything modified by in this transformation, and therefore included in the modified keys,
// should have the same epoch as the CM itself. Anything not modified now must have a strictly
// earlier epoch
- for (MetadataKey key : Iterables.concat(MetadataKeys.CORE_METADATA, transformed.metadata.extensions.keySet()))
+ for (MetadataKey key : Iterables.concat(MetadataKeys.CORE_METADATA.keySet(), transformed.metadata.extensions.keySet()))
{
MetadataValue<?> value = valueFor(key, transformed.metadata);
if (transformed.modifiedKeys.contains(key))
@@ -287,7 +287,7 @@ public class ClusterMetadataTransformationTest
private static MetadataValue<?> valueFor(MetadataKey key, ClusterMetadata metadata)
{
- if (!MetadataKeys.CORE_METADATA.contains(key))
+ if (!MetadataKeys.CORE_METADATA.containsKey(key))
{
assert key instanceof ExtensionKey<?,?>;
return metadata.extensions.get((ExtensionKey<?, ?>)key);
diff --git a/test/unit/org/apache/cassandra/tcm/ValidatingClusterMetadataService.java b/test/unit/org/apache/cassandra/tcm/ValidatingClusterMetadataService.java
new file mode 100644
index 0000000000..fd37346aa0
--- /dev/null
+++ b/test/unit/org/apache/cassandra/tcm/ValidatingClusterMetadataService.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tcm;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
+import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializers;
+import org.apache.cassandra.tcm.serialization.Version;
+import org.assertj.core.api.Assertions;
+
+public class ValidatingClusterMetadataService extends StubClusterMetadataService
+{
+ private final List<Version> supportedVersions;
+
+ private ValidatingClusterMetadataService(List<Version> supportedVersions)
+ {
+ super(new ClusterMetadata(safeGetPartitioner()));
+ this.supportedVersions = supportedVersions;
+ }
+
+ public static ValidatingClusterMetadataService createAndRegister(Version minVersion)
+ {
+ return createAndRegister(minVersion.greaterThanOrEqual());
+ }
+
+ public static ValidatingClusterMetadataService createAndRegister(List<Version> supportedVersions)
+ {
+ ValidatingClusterMetadataService cms = new ValidatingClusterMetadataService(supportedVersions);
+
+ ClusterMetadataService.unsetInstance();
+ ClusterMetadataService.setInstance(cms);
+ return cms;
+ }
+
+ private static IPartitioner safeGetPartitioner()
+ {
+ IPartitioner partitioner = DatabaseDescriptor.getPartitioner();
+ return partitioner == null ? Murmur3Partitioner.instance : partitioner;
+ }
+
+ private <In, Out> void testSerde(AsymmetricMetadataSerializer<In, Out> serializer, In input)
+ {
+ for (Version version : supportedVersions)
+ {
+ try (DataOutputBuffer buffer = DataOutputBuffer.scratchBuffer.get())
+ {
+ AsymmetricMetadataSerializers.testSerde(buffer, serializer, input, version);
+ }
+ catch (IOException e)
+ {
+ throw new AssertionError(String.format("Serde error for version=%s; input=%s", version, input), e);
+ }
+ }
+ }
+
+ @Override
+ protected Transformation.Result execute(Transformation transform)
+ {
+ Transformation.Result result = super.execute(transform);
+ if (result.isSuccess())
+ {
+ Transformation.Success success = result.success();
+ Assertions.assertThat(success.affectedMetadata)
+ .describedAs("Affected Metadata keys do not match")
+ .isEqualTo(MetadataKeys.diffKeys(metadata(), success.metadata));
+ }
+ return result;
+ }
+
+ @Override
+ public <T1> T1 commit(Transformation transform, CommitSuccessHandler<T1> onSuccess, CommitFailureHandler<T1> onFailure)
+ {
+ testSerde(transform.kind().serializer(), transform);
+ return super.commit(transform, onSuccess, onFailure);
+ }
+
+ @Override
+ public void setMetadata(ClusterMetadata metadata)
+ {
+ if (!metadata.epoch.equals(metadata().epoch.nextEpoch()))
+ throw new AssertionError("Epochs were not sequential: expected " + metadata().epoch.nextEpoch() + " but given " + metadata.epoch);
+ testSerde(ClusterMetadata.serializer, metadata);
+ super.setMetadata(metadata);
+ }
+}
diff --git a/test/unit/org/apache/cassandra/tcm/log/LogListenerNotificationTest.java b/test/unit/org/apache/cassandra/tcm/log/LogListenerNotificationTest.java
index 59aafa37e0..55df4e868e 100644
--- a/test/unit/org/apache/cassandra/tcm/log/LogListenerNotificationTest.java
+++ b/test/unit/org/apache/cassandra/tcm/log/LogListenerNotificationTest.java
@@ -111,7 +111,7 @@ public class LogListenerNotificationTest
static Set<MetadataKey> affectedMetadata(Random random)
{
- List<MetadataKey> src = new ArrayList<>(CORE_METADATA);
+ List<MetadataKey> src = new ArrayList<>(CORE_METADATA.keySet());
int required = random.nextInt(src.size());
Set<MetadataKey> keys = new HashSet<>();
while (keys.size() < required)
diff --git a/test/unit/org/apache/cassandra/tcm/sequences/DropAccordTableTest.java b/test/unit/org/apache/cassandra/tcm/sequences/DropAccordTableTest.java
new file mode 100644
index 0000000000..8b18368809
--- /dev/null
+++ b/test/unit/org/apache/cassandra/tcm/sequences/DropAccordTableTest.java
@@ -0,0 +1,223 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tcm.sequences;
+
+import java.util.TreeSet;
+import java.util.stream.Stream;
+
+import org.junit.Test;
+
+import accord.utils.Gen;
+import accord.utils.Property;
+import accord.utils.Property.Command;
+import accord.utils.RandomSource;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.locator.MetaStrategy;
+import org.apache.cassandra.schema.DistributedSchema;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.schema.KeyspaceParams;
+import org.apache.cassandra.schema.Keyspaces;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.schema.Types;
+import org.apache.cassandra.service.consensus.TransactionalMode;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.ClusterMetadataService;
+import org.apache.cassandra.tcm.MultiStepOperation;
+import org.apache.cassandra.tcm.StubClusterMetadataService;
+import org.apache.cassandra.tcm.ValidatingClusterMetadataService;
+import org.apache.cassandra.tcm.serialization.Version;
+import org.apache.cassandra.tcm.transformations.PrepareDropAccordTable;
+import org.apache.cassandra.tcm.sequences.DropAccordTable.TableReference;
+import org.apache.cassandra.utils.AbstractTypeGenerators;
+import org.apache.cassandra.utils.CassandraGenerators;
+import org.apache.cassandra.utils.CassandraGenerators.TableMetadataBuilder;
+import org.apache.cassandra.utils.Generators;
+import org.assertj.core.api.Assertions;
+import org.quicktheories.generators.SourceDSL;
+
+import static accord.utils.Property.commands;
+import static accord.utils.Property.qt;
+import static accord.utils.Property.stateful;
+import static org.apache.cassandra.utils.CassandraGenerators.TABLE_ID_GEN;
+
+public class DropAccordTableTest
+{
+ static
+ {
+ DatabaseDescriptor.clientInitialization();
+ }
+
+ private static final TransactionalMode[] ACCORD_ENABLED_MODES = Stream.of(TransactionalMode.values())
+ .filter(t -> t.accordIsEnabled)
+ .toArray(TransactionalMode[]::new);
+
+ private static final Gen<TableMetadata> TABLE_GEN = Generators.toGen(defaultTableMetadataBuilder().build());
+
+ private static TableMetadataBuilder defaultTableMetadataBuilder()
+ {
+ return new TableMetadataBuilder()
+ .withUseCounter(false)
+ .withPartitioner(Murmur3Partitioner.instance)
+ .withTransactionalMode(SourceDSL.arbitrary().pick(ACCORD_ENABLED_MODES));
+ }
+
+ @Test
+ public void e2e()
+ {
+ qt().check(rs -> {
+ ValidatingClusterMetadataService cms = createCMS();
+ TableMetadata metadata = TABLE_GEN.next(rs);
+ addTable(cms, metadata); // hack this table into the schema...
+
+ TableReference table = TableReference.from(metadata);
+
+ cms.commit(new PrepareDropAccordTable(table));
+
+ // This is only here because "applyTo" is not touched without it...
+ for (KeyspaceMetadata ks : cms.metadata().schema.getKeyspaces())
+ cms.metadata().writePlacementAllSettled(ks);
+
+ Assertions.assertThat(cms.metadata().inProgressSequences.isEmpty()).isFalse();
+ InProgressSequences.finishInProgressSequences(table);
+ Assertions.assertThat(cms.metadata().inProgressSequences.isEmpty()).isTrue();
+
+ // table is dropped
+ Assertions.assertThat(cms.metadata().schema.getTableMetadata(metadata.id)).isNull();
+ });
+ }
+
+ @Test
+ public void multi()
+ {
+ stateful().withExamples(50).withSteps(500).check(commands(() -> State::new)
+ .destroyState(DropAccordTableTest::validate)
+ .add(DropAccordTableTest::addTable)
+ .addIf(s -> !s.aliveTables.isEmpty(), DropAccordTableTest::dropTable)
+ .addIf(s -> !s.cms.metadata().inProgressSequences.isEmpty(), DropAccordTableTest::inProgressSequences)
+ .build());
+ }
+
+ private static void validate(State state)
+ {
+ while (!state.cms.metadata().inProgressSequences.isEmpty())
+ {
+ for (MultiStepOperation<?> opt : state.cms.metadata().inProgressSequences)
+ InProgressSequences.resume(opt);
+ }
+ // all tables are dropped, unless they were never dropped
+ Keyspaces keyspaces = state.cms.metadata().schema.getKeyspaces();
+ for (KeyspaceMetadata k : keyspaces)
+ {
+ if (k.tables.size() == 0) continue;
+ if (k.replicationStrategy instanceof MetaStrategy) continue;
+ for (TableMetadata t : k.tables)
+ {
+ Assertions.assertThat(t.params.pendingDrop).isFalse();
+ Assertions.assertThat(state.aliveTables).contains(t.id);
+ }
+ }
+ }
+
+ private static Command<State, Void, ?> addTable(RandomSource rs, State state)
+ {
+ TableMetadata metadata = Generators.toGen(defaultTableMetadataBuilder()
+ .withKeyspaceName(CassandraGenerators.KEYSPACE_NAME_GEN.assuming(name -> !state.cms.metadata().schema.getKeyspaces().containsKeyspace(name)))
+ .withTableId(TABLE_ID_GEN.assuming(id -> state.cms.metadata().schema.getTableMetadata(id) == null))
+ // other tests better cover serialization so can speed up tests by only doing primitive types
+ .withDefaultTypeGen(CassandraGenerators.TableMetadataBuilder.defaultTypeGen().withTypeKinds(AbstractTypeGenerators.TypeKind.PRIMITIVE))
+ .build())
+ .next(rs);
+ return new Property.SimpleCommand<>("Add Table " + metadata, s2 -> {
+ addTable(s2.cms, metadata);
+ s2.aliveTables.add(metadata.id);
+ });
+ }
+
+ private static Command<State, Void, ?> dropTable(RandomSource rs, State state)
+ {
+ TableId id = rs.pickOrderedSet(state.aliveTables);
+ TableMetadata metadata = state.cms.metadata().schema.getTableMetadata(id);
+ return new Property.SimpleCommand<>("Drop Table " + metadata, s2 -> {
+ TableReference table = TableReference.from(metadata);
+
+ s2.cms.commit(new PrepareDropAccordTable(table));
+ s2.aliveTables.remove(id);
+ });
+ }
+
+ private static Command<State, Void, ?> inProgressSequences(RandomSource rs, State state)
+ {
+ ClusterMetadata current = state.cms.metadata();
+ TreeSet<TableReference> pending = new TreeSet<>();
+ for (MultiStepOperation<?> opt : current.inProgressSequences)
+ {
+ if (!(opt instanceof DropAccordTable)) throw new AssertionError("Only DropAccordTable should exist in this test; found " + opt);
+ pending.add(((DropAccordTable) opt).table);
+ }
+ TableReference ref = rs.pickOrderedSet(pending);
+ MultiStepOperation<?> seq = current.inProgressSequences.get(ref);
+ Assertions.assertThat(seq).isNotNull();
+ return new Property.SimpleCommand<>("Progress for " + ref + ": " + seq.nextStep(), s2 -> InProgressSequences.resume(seq));
+ }
+
+ public static class State
+ {
+ private final StubClusterMetadataService cms;
+ private final TreeSet<TableId> aliveTables = new TreeSet<>();
+
+ public State(RandomSource rs)
+ {
+ // With validation enabled the test runtime is dominated by serialization checks, so enable them rarely
+ // just so tests do run with them, but the whole test runtime isn't serde testing.
+ if (rs.decide(0.01))
+ {
+ cms = ValidatingClusterMetadataService.createAndRegister(Version.MIN_ACCORD_VERSION);
+ }
+ else
+ {
+ cms = StubClusterMetadataService.forTesting(new ClusterMetadata(Murmur3Partitioner.instance));
+ ClusterMetadataService.unsetInstance();
+ ClusterMetadataService.setInstance(cms);
+ }
+ }
+ }
+
+ private static ValidatingClusterMetadataService createCMS()
+ {
+ return ValidatingClusterMetadataService.createAndRegister(Version.MIN_ACCORD_VERSION);
+ }
+
+ private static void addTable(StubClusterMetadataService cms, TableMetadata table)
+ {
+ class Ref { Types types;}
+ // first mock out a keyspace
+ ClusterMetadata prev = cms.metadata();
+ KeyspaceMetadata schema = KeyspaceMetadata.create(table.keyspace, KeyspaceParams.simple(3));
+ Ref ref = new Ref();
+ ref.types = schema.types;
+ CassandraGenerators.visitUDTs(table, udt -> ref.types = ref.types.with(udt.unfreeze()));
+ schema = schema.withSwapped(ref.types);
+ schema = schema.withSwapped(schema.tables.with(table));
+ Keyspaces keyspaces = prev.schema.getKeyspaces().withAddedOrUpdated(schema);
+ ClusterMetadata metadata = prev.transformer().with(new DistributedSchema(keyspaces)).build().metadata;
+ cms.setMetadata(metadata);
+ }
+}
\ No newline at end of file
diff --git a/test/unit/org/apache/cassandra/tcm/transformations/AccordMarkRejoiningTest.java b/test/unit/org/apache/cassandra/tcm/transformations/AccordMarkRejoiningTest.java
index 032c8dd1ec..32999a8aae 100644
--- a/test/unit/org/apache/cassandra/tcm/transformations/AccordMarkRejoiningTest.java
+++ b/test/unit/org/apache/cassandra/tcm/transformations/AccordMarkRejoiningTest.java
@@ -35,7 +35,7 @@ public class AccordMarkRejoiningTest
public void shouldSerializeEmpty() throws IOException
{
DataOutputBuffer buffer = new DataOutputBuffer();
- AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkRejoining.serializer, new AccordMarkRejoining(Collections.emptySet()), Version.V2);
+ AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkRejoining.serializer, new AccordMarkRejoining(Collections.emptySet()), Version.MIN_ACCORD_VERSION);
}
@Test
@@ -43,7 +43,7 @@ public class AccordMarkRejoiningTest
{
DataOutputBuffer buffer = new DataOutputBuffer();
AccordMarkRejoining markStale = new AccordMarkRejoining(Collections.singleton(NodeId.fromString("1")));
- AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkRejoining.serializer, markStale, Version.V2);
+ AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkRejoining.serializer, markStale, Version.MIN_ACCORD_VERSION);
}
@Test
@@ -51,6 +51,6 @@ public class AccordMarkRejoiningTest
{
DataOutputBuffer buffer = new DataOutputBuffer();
AccordMarkRejoining markStale = new AccordMarkRejoining(ImmutableSet.of(NodeId.fromString("1"), NodeId.fromString("2")));
- AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkRejoining.serializer, markStale, Version.V2);
+ AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkRejoining.serializer, markStale, Version.MIN_ACCORD_VERSION);
}
}
diff --git a/test/unit/org/apache/cassandra/tcm/transformations/AccordMarkStaleTest.java b/test/unit/org/apache/cassandra/tcm/transformations/AccordMarkStaleTest.java
index d794b3a2a9..baa4936a07 100644
--- a/test/unit/org/apache/cassandra/tcm/transformations/AccordMarkStaleTest.java
+++ b/test/unit/org/apache/cassandra/tcm/transformations/AccordMarkStaleTest.java
@@ -35,7 +35,7 @@ public class AccordMarkStaleTest
public void shouldSerializeEmpty() throws IOException
{
DataOutputBuffer buffer = new DataOutputBuffer();
- AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkStale.serializer, new AccordMarkStale(Collections.emptySet()), Version.V2);
+ AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkStale.serializer, new AccordMarkStale(Collections.emptySet()), Version.MIN_ACCORD_VERSION);
}
@Test
@@ -43,7 +43,7 @@ public class AccordMarkStaleTest
{
DataOutputBuffer buffer = new DataOutputBuffer();
AccordMarkStale markStale = new AccordMarkStale(Collections.singleton(NodeId.fromString("1")));
- AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkStale.serializer, markStale, Version.V2);
+ AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkStale.serializer, markStale, Version.MIN_ACCORD_VERSION);
}
@Test
@@ -51,6 +51,6 @@ public class AccordMarkStaleTest
{
DataOutputBuffer buffer = new DataOutputBuffer();
AccordMarkStale markStale = new AccordMarkStale(ImmutableSet.of(NodeId.fromString("1"), NodeId.fromString("2")));
- AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkStale.serializer, markStale, Version.V2);
+ AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkStale.serializer, markStale, Version.MIN_ACCORD_VERSION);
}
}
diff --git a/test/unit/org/apache/cassandra/utils/CassandraGenerators.java b/test/unit/org/apache/cassandra/utils/CassandraGenerators.java
index 3abae215e5..8f3ffa2439 100644
--- a/test/unit/org/apache/cassandra/utils/CassandraGenerators.java
+++ b/test/unit/org/apache/cassandra/utils/CassandraGenerators.java
@@ -28,6 +28,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Deque;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
@@ -37,6 +38,7 @@ import java.util.Objects;
import java.util.Set;
import java.util.TreeMap;
import java.util.UUID;
+import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -51,6 +53,7 @@ import com.google.common.collect.Sets;
import org.apache.commons.lang3.builder.MultilineRecursiveToStringStyle;
import org.apache.commons.lang3.builder.ReflectionToStringBuilder;
+import accord.local.Node;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.ColumnIdentifier;
import org.apache.cassandra.cql3.Duration;
@@ -65,6 +68,7 @@ import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.db.marshal.CompositeType;
import org.apache.cassandra.db.marshal.CounterColumnType;
import org.apache.cassandra.db.marshal.EmptyType;
+import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.rows.Cell;
import org.apache.cassandra.db.marshal.UserType;
import org.apache.cassandra.dht.ByteOrderedPartitioner;
@@ -97,6 +101,7 @@ import org.apache.cassandra.net.Verb;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.KeyspaceParams;
+import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.schema.MemtableParams;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.schema.TableId;
@@ -109,6 +114,20 @@ import org.apache.cassandra.schema.Views;
import org.apache.cassandra.service.accord.fastpath.FastPathStrategy;
import org.apache.cassandra.service.consensus.TransactionalMode;
import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.service.accord.AccordFastPath;
+import org.apache.cassandra.service.accord.AccordStaleReplicas;
+import org.apache.cassandra.service.accord.fastpath.InheritKeyspaceFastPathStrategy;
+import org.apache.cassandra.service.accord.fastpath.ParameterizedFastPathStrategy;
+import org.apache.cassandra.service.accord.fastpath.SimpleFastPathStrategy;
+import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState;
+import org.apache.cassandra.tcm.Epoch;
+import org.apache.cassandra.tcm.extensions.ExtensionKey;
+import org.apache.cassandra.tcm.extensions.ExtensionValue;
+import org.apache.cassandra.tcm.membership.Directory;
+import org.apache.cassandra.tcm.ownership.DataPlacements;
+import org.apache.cassandra.tcm.ownership.TokenMap;
+import org.apache.cassandra.tcm.sequences.InProgressSequences;
+import org.apache.cassandra.tcm.sequences.LockedRanges;
import org.apache.cassandra.utils.AbstractTypeGenerators.TypeGenBuilder;
import org.apache.cassandra.utils.AbstractTypeGenerators.ValueDomain;
import org.quicktheories.core.Gen;
@@ -414,6 +433,109 @@ public final class CassandraGenerators
}
}
+ public static class TableParamsBuilder
+ {
+ @Nullable
+ private Gen<String> memtableKeyGen = null;
+ @Nullable
+ private Gen<TransactionalMode> transactionalMode = null;
+ @Nullable
+ private Gen<FastPathStrategy> fastPathStrategy = null;
+
+ public TableParamsBuilder withKnownMemtables()
+ {
+ Set<String> known = MemtableParams.knownDefinitions();
+ // for testing reason, some invalid types are added; filter out
+ List<String> valid = known.stream().filter(name -> !name.startsWith("test_")).collect(Collectors.toList());
+ memtableKeyGen = SourceDSL.arbitrary().pick(valid);
+ return this;
+ }
+
+ public TableParamsBuilder withTransactionalMode(Gen<TransactionalMode> transactionalMode)
+ {
+ this.transactionalMode = transactionalMode;
+ return this;
+ }
+
+ public TableParamsBuilder withTransactionalMode()
+ {
+ return withTransactionalMode(SourceDSL.arbitrary().enumValues(TransactionalMode.class));
+ }
+
+ public TableParamsBuilder withTransactionalMode(TransactionalMode transactionalMode)
+ {
+ return withTransactionalMode(SourceDSL.arbitrary().constant(transactionalMode));
+ }
+
+ public TableParamsBuilder withFastPathStrategy()
+ {
+ fastPathStrategy = rnd -> {
+ FastPathStrategy.Kind kind = SourceDSL.arbitrary().enumValues(FastPathStrategy.Kind.class).generate(rnd);
+ switch (kind)
+ {
+ case SIMPLE:
+ return SimpleFastPathStrategy.instance;
+ case INHERIT_KEYSPACE:
+ return InheritKeyspaceFastPathStrategy.instance;
+ case PARAMETERIZED:
+ {
+ Map<String, String> map = new HashMap<>();
+ int size = SourceDSL.integers().between(1, Integer.MAX_VALUE).generate(rnd);
+ map.put(ParameterizedFastPathStrategy.SIZE, Integer.toString(size));
+ Set<String> names = new HashSet<>();
+ Gen<String> nameGen = SourceDSL.strings().allPossible().ofLengthBetween(1, 10).assuming(s -> !s.trim().isEmpty());
+ int numNames = SourceDSL.integers().between(1, 10).generate(rnd);
+ for (int i = 0; i < numNames; i++)
+ {
+ while (!names.add(nameGen.generate(rnd)))
+ {
+ }
+ }
+ List<String> sortedNames = new ArrayList<>(names);
+ sortedNames.sort(Comparator.naturalOrder());
+ List<String> dcs = new ArrayList<>(names.size());
+ boolean auto = SourceDSL.booleans().all().generate(rnd);
+ if (auto)
+ {
+ dcs.addAll(sortedNames);
+ }
+ else
+ {
+ for (String name : sortedNames)
+ {
+ int weight = SourceDSL.integers().between(0, 10).generate(rnd);
+ dcs.add(name + ":" + weight);
+ }
+ }
+ // str: dcFormat(,dcFormat)*
+ // dcFormat: name | weight
+ // weight: int: >= 0
+ // note: can't mix auto and user defined weight; need one or the other. Names must be unique
+ map.put(ParameterizedFastPathStrategy.DCS, String.join(",", dcs));
+ return ParameterizedFastPathStrategy.fromMap(map);
+ }
+ default:
+ throw new UnsupportedOperationException(kind.name());
+ }
+ };
+ return this;
+ }
+
+ public Gen<TableParams> build()
+ {
+ return rnd -> {
+ TableParams.Builder params = TableParams.builder();
+ if (memtableKeyGen != null)
+ params.memtable(MemtableParams.get(memtableKeyGen.generate(rnd)));
+ if (transactionalMode != null)
+ params.transactionalMode(transactionalMode.generate(rnd));
+ if (fastPathStrategy != null)
+ params.fastPath(fastPathStrategy.generate(rnd));
+ return params.build();
+ };
+ }
+ }
+
public static TableMetadataBuilder regularTable()
{
return new TableMetadataBuilder()
@@ -434,11 +556,10 @@ public final class CassandraGenerators
private Gen<Integer> numClusteringColumnsGen = SourceDSL.integers().between(1, 2);
private Gen<Integer> numRegularColumnsGen = SourceDSL.integers().between(1, 5);
private Gen<Integer> numStaticColumnsGen = SourceDSL.integers().between(0, 2);
- private Gen<String> memtableKeyGen = null;
- @Nullable
- private Gen<TransactionalMode> transactionalMode = null;
@Nullable
private ColumnNameGen columnNameGen = null;
+ private TableParamsBuilder paramsBuilder = new TableParamsBuilder();
+ private Gen<IPartitioner> partitionerGen = partitioners();
public static TypeGenBuilder defaultTypeGen()
{
@@ -455,7 +576,7 @@ public final class CassandraGenerators
switch (kind)
{
case PARTITION_KEY: return "pk" + offset;
- case CLUSTERING: return "ck" + offset;
+ case CLUSTERING: return "ck" + offset;
case STATIC: return "s" + offset;
case REGULAR: return "v" + offset;
default: throw new UnsupportedOperationException("Unknown kind: " + kind);
@@ -464,6 +585,22 @@ public final class CassandraGenerators
return this;
}
+ public TableMetadataBuilder withPartitioner(Gen<IPartitioner> partitionerGen)
+ {
+ this.partitionerGen = Objects.requireNonNull(partitionerGen);
+ return this;
+ }
+
+ public TableMetadataBuilder withPartitioner(IPartitioner partitioner)
+ {
+ return withPartitioner(i -> partitioner);
+ }
+
+ public TableMetadataBuilder withUseCounter(boolean useCounter)
+ {
+ return withUseCounter(ignore -> useCounter);
+ }
+
public TableMetadataBuilder withUseCounter(Gen<Boolean> useCounter)
{
this.useCounter = Objects.requireNonNull(useCounter);
@@ -472,21 +609,19 @@ public final class CassandraGenerators
public TableMetadataBuilder withTransactionalMode(Gen<TransactionalMode> transactionalMode)
{
- this.transactionalMode = transactionalMode;
+ paramsBuilder.withTransactionalMode(transactionalMode);
return this;
}
public TableMetadataBuilder withTransactionalMode(TransactionalMode transactionalMode)
{
- return withTransactionalMode(SourceDSL.arbitrary().constant(transactionalMode));
+ paramsBuilder.withTransactionalMode(transactionalMode);
+ return this;
}
public TableMetadataBuilder withKnownMemtables()
{
- Set<String> known = MemtableParams.knownDefinitions();
- // for testing reason, some invalid types are added; filter out
- List<String> valid = known.stream().filter(name -> !name.startsWith("test_")).collect(Collectors.toList());
- memtableKeyGen = SourceDSL.arbitrary().pick(valid);
+ paramsBuilder.withKnownMemtables();
return this;
}
@@ -640,22 +775,16 @@ public final class CassandraGenerators
try
{
String tableName = tableNameGen.generate(rnd);
- TableParams.Builder params = TableParams.builder();
- if (memtableKeyGen != null)
- params.memtable(MemtableParams.get(memtableKeyGen.generate(rnd)));
- if (transactionalMode != null)
- params.transactionalMode(transactionalMode.generate(rnd));
+ TableParams params = paramsBuilder.build().generate(rnd);
boolean isCounter = useCounter.generate(rnd);
TableMetadata.Builder builder = TableMetadata.builder(ks, tableName, tableIdGen.generate(rnd))
- .partitioner(partitioners().generate(rnd))
- .kind(tableKindGen.generate(rnd))
- .isCounter(isCounter)
- .params(params.build());
+ .partitioner(partitionerGen.generate(rnd))
+ .kind(tableKindGen.generate(rnd))
+ .isCounter(isCounter)
+ .params(params);
int numPartitionColumns = numPartitionColumnsGen.generate(rnd);
int numClusteringColumns = numClusteringColumnsGen.generate(rnd);
- int numRegularColumns = numRegularColumnsGen.generate(rnd);
- int numStaticColumns = numStaticColumnsGen.generate(rnd);
ColumnNameGen nameGen;
if (columnNameGen != null)
@@ -685,6 +814,8 @@ public final class CassandraGenerators
}
else
{
+ int numRegularColumns = numRegularColumnsGen.generate(rnd);
+ int numStaticColumns = numStaticColumnsGen.generate(rnd);
for (int i = 0; i < numStaticColumns; i++)
builder.addColumn(createColumnDefinition(ks, tableName, ColumnMetadata.Kind.STATIC, i, nameGen, staticColTypeGen, rnd));
for (int i = 0; i < numRegularColumns; i++)
@@ -942,13 +1073,11 @@ public final class CassandraGenerators
};
}
- public static Gen<Token> localCompositePrefixPartitionerToken()
+ public static Gen<Token> localCompositePrefixPartitionerToken(LocalCompositePrefixPartitioner partitioner)
{
- var lpGen = localCompositePrefixPartitioner();
return rs -> {
- var lp = lpGen.generate(rs);
- var bytes = AbstractTypeGenerators.getTypeSupport(lp.getTokenValidator()).bytesGen();
- return lp.getToken(bytes.generate(rs));
+ var bytes = AbstractTypeGenerators.getTypeSupport(partitioner.getTokenValidator()).bytesGen();
+ return partitioner.getToken(bytes.generate(rs));
};
}
@@ -1033,6 +1162,24 @@ public final class CassandraGenerators
.flatMap(SupportedPartitioners::partitioner);
}
+ /**
+ * For {@link LocalPartitioner} it can have a very complex type which can lead to generating data larger than
+ * allowed in a primary key. If a test needs to filter out those cases, can just
+ * {@code .map(CassandraGenerators::simplify)} to resolve.
+ */
+ public static IPartitioner simplify(IPartitioner partitioner)
+ {
+ // serializers require tokens to fit within 1 << 16, but that makes the test flakey when LocalPartitioner with a nested type is found...
+ if (!(partitioner instanceof LocalPartitioner)) return partitioner;
+ if (!shouldSimplify(partitioner.getTokenValidator())) return partitioner;
+ return new LocalPartitioner(Int32Type.instance);
+ }
+
+ private static boolean shouldSimplify(AbstractType<?> type)
+ {
+ return AbstractTypeGenerators.contains(type, t -> t.isCollection());
+ }
+
public static Gen<Token> token()
{
return partitioners().flatMap(CassandraGenerators::token);
@@ -1043,7 +1190,7 @@ public final class CassandraGenerators
if (partitioner instanceof Murmur3Partitioner) return murmurToken();
if (partitioner instanceof ByteOrderedPartitioner) return byteOrderToken();
if (partitioner instanceof RandomPartitioner) return randomPartitionerToken();
- if (partitioner instanceof LocalCompositePrefixPartitioner) return localCompositePrefixPartitionerToken();
+ if (partitioner instanceof LocalCompositePrefixPartitioner) return localCompositePrefixPartitionerToken((LocalCompositePrefixPartitioner) partitioner);
if (partitioner instanceof LocalPartitioner) return localPartitionerToken((LocalPartitioner) partitioner);
if (partitioner instanceof OrderPreservingPartitioner) return orderPreservingToken();
throw new UnsupportedOperationException("Unsupported partitioner: " + partitioner.getClass());
@@ -1333,4 +1480,82 @@ public final class CassandraGenerators
return types;
}
}
+
+ private enum EpochConstants { FIRST, EMPTY, UPGRADE_STARTUP, UPGRADE_GOSSIP}
+ public static Gen<Epoch> epochs()
+ {
+ return rnd -> {
+ if (SourceDSL.booleans().all().generate(rnd))
+ {
+ switch (SourceDSL.arbitrary().enumValues(EpochConstants.class).generate(rnd))
+ {
+ case FIRST: return Epoch.FIRST;
+ case EMPTY: return Epoch.EMPTY;
+ case UPGRADE_STARTUP: return Epoch.UPGRADE_STARTUP;
+ case UPGRADE_GOSSIP: return Epoch.UPGRADE_GOSSIP;
+ default: throw new UnsupportedOperationException();
+ }
+ }
+
+ return Epoch.create(SourceDSL.longs().between(2, Long.MAX_VALUE).generate(rnd));
+ };
+ }
+
+ public static Gen<Node.Id> accordNodeId()
+ {
+ return SourceDSL.integers().between(0, Integer.MAX_VALUE).map(Node.Id::new);
+ }
+
+ public static Gen<AccordStaleReplicas> accordStaleReplicas()
+ {
+ Gen<Set<Node.Id>> staleIdsGen = Generators.set(accordNodeId(), SourceDSL.integers().between(0, 10));
+ Gen<Epoch> epochGen = epochs();
+ return rnd -> new AccordStaleReplicas(staleIdsGen.generate(rnd), epochGen.generate(rnd));
+ }
+
+ public static Gen<AccordFastPath> accordFastPath()
+ {
+ Gen<List<Node.Id>> nodesGen = Generators.uniqueList(accordNodeId(), SourceDSL.integers().between(0, 10));
+ Gen<AccordFastPath.Status> statusGen = SourceDSL.arbitrary().enumValues(AccordFastPath.Status.class);
+ Gen<Long> updateTimeMillis = TIMESTAMP_NANOS.map(TimeUnit.NANOSECONDS::toMillis);
+ Gen<Long> updateDelayMillis = SourceDSL.longs().between(0, TimeUnit.HOURS.toMillis(2));
+ return rnd -> {
+ AccordFastPath accum = AccordFastPath.EMPTY;
+ for (Node.Id node : nodesGen.generate(rnd))
+ {
+ AccordFastPath.Status status = statusGen.generate(rnd);
+ // can't add a NORMAL node that doesn't exist, it must be ab-NORMAL first...
+ if (status == AccordFastPath.Status.NORMAL)
+ accum = accum.withNodeStatusSince(node, AccordFastPath.Status.UNAVAILABLE, 0, 0);
+ accum = accum.withNodeStatusSince(node, status, updateTimeMillis.generate(rnd), updateDelayMillis.generate(rnd));
+ }
+ return accum;
+ };
+ }
+
+ public static class ClusterMetadataBuilder
+ {
+ private Gen<Epoch> epochGen = epochs();
+ private Gen<IPartitioner> partitionerGen = nonLocalPartitioners();
+ private Gen<AccordStaleReplicas> accordStaleReplicasGen = accordStaleReplicas();
+ private Gen<AccordFastPath> accordFastPathGen = accordFastPath();
+ public Gen<ClusterMetadata> build()
+ {
+ return rnd -> {
+ Epoch epoch = epochGen.generate(rnd);
+ IPartitioner partitioner = partitionerGen.generate(rnd);
+ Directory directory = Directory.EMPTY;
+ DistributedSchema schema = DistributedSchema.first(directory.knownDatacenters());
+ TokenMap tokenMap = new TokenMap(partitioner);
+ DataPlacements placements = DataPlacements.EMPTY;
+ AccordFastPath accordFastPath = accordFastPathGen.generate(rnd);
+ LockedRanges lockedRanges = LockedRanges.EMPTY;
+ InProgressSequences inProgressSequences = InProgressSequences.EMPTY;
+ ConsensusMigrationState consensusMigrationState = ConsensusMigrationState.EMPTY;
+ Map<ExtensionKey<?, ?>, ExtensionValue<?>> extensions = ImmutableMap.of();
+ AccordStaleReplicas accordStaleReplicas = accordStaleReplicasGen.generate(rnd);
+ return new ClusterMetadata(epoch, partitioner, schema, directory, tokenMap, placements, accordFastPath, lockedRanges, inProgressSequences, consensusMigrationState, extensions, accordStaleReplicas);
+ };
+ }
+ }
}
diff --git a/test/unit/org/apache/cassandra/utils/CassandraGeneratorsTest.java b/test/unit/org/apache/cassandra/utils/CassandraGeneratorsTest.java
index 2a31c135cf..856e9a0780 100644
--- a/test/unit/org/apache/cassandra/utils/CassandraGeneratorsTest.java
+++ b/test/unit/org/apache/cassandra/utils/CassandraGeneratorsTest.java
@@ -47,7 +47,7 @@ public class CassandraGeneratorsTest
@Test
public void partitionerToToken()
{
- qt().forAll(Gens.random(), toGen(CassandraGenerators.partitioners()))
+ qt().forAll(Gens.random(), toGen(CassandraGenerators.partitioners().map(CassandraGenerators::simplify)))
.check((rs, p) -> Assertions.assertThat(toGen(CassandraGenerators.token(p)).next(rs)).isNotNull());
}
diff --git a/test/unit/org/apache/cassandra/utils/Generators.java b/test/unit/org/apache/cassandra/utils/Generators.java
index cac1248c86..7c2ae41d09 100644
--- a/test/unit/org/apache/cassandra/utils/Generators.java
+++ b/test/unit/org/apache/cassandra/utils/Generators.java
@@ -26,8 +26,11 @@ import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
import java.time.ZoneOffset;
import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.Comparator;
import java.util.Date;
import java.util.HashSet;
+import java.util.List;
import java.util.Random;
import java.util.Set;
import java.util.UUID;
@@ -485,6 +488,28 @@ public final class Generators
return filter(gen, dedup::add);
}
+ public static <T> Gen<Set<T>> set(Gen<T> gen, Gen<Integer> sizeGen)
+ {
+ return rnd -> {
+ Set<T> set = new HashSet<>();
+ int size = sizeGen.generate(rnd);
+ for (int i = 0; i < size; i++)
+ {
+ while (!set.add(gen.generate(rnd))) {}
+ }
+ return set;
+ };
+ }
+
+ public static <T extends Comparable<T>> Gen<List<T>> uniqueList(Gen<T> gen, Gen<Integer> sizeGen)
+ {
+ return set(gen, sizeGen).map(t -> {
+ List<T> list = new ArrayList<>(t);
+ list.sort(Comparator.naturalOrder());
+ return list;
+ });
+ }
+
public static <T> Gen<T> cached(Gen<T> gen)
{
Object cacheMissed = new Object();
diff --git a/test/unit/org/apache/cassandra/utils/SimulatedMiniCluster.java b/test/unit/org/apache/cassandra/utils/SimulatedMiniCluster.java
index 279059b476..a2ed968503 100644
--- a/test/unit/org/apache/cassandra/utils/SimulatedMiniCluster.java
+++ b/test/unit/org/apache/cassandra/utils/SimulatedMiniCluster.java
@@ -86,6 +86,7 @@ import org.apache.cassandra.tcm.membership.Directory;
import org.apache.cassandra.tcm.membership.Location;
import org.apache.cassandra.tcm.membership.NodeAddresses;
import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.tcm.membership.NodeVersion;
import org.apache.cassandra.tcm.ownership.UniformRangePlacement;
import org.apache.cassandra.tcm.transformations.PrepareJoin;
import org.mockito.Mockito;
@@ -231,9 +232,10 @@ public class SimulatedMiniCluster
}
else
{
- notifyMetadataChange(current.forceEpoch(current.epoch.nextEpoch())
- .withDirectory(current.directory.with(new NodeAddresses(node.hostId, node.broadcastAddressAndPort, node.broadcastAddressAndPort, node.broadcastAddressAndPort),
- new Location(node.dc, node.rack))));
+ notifyMetadataChange(current.transformer().register(new NodeAddresses(node.hostId, node.broadcastAddressAndPort, node.broadcastAddressAndPort, node.broadcastAddressAndPort),
+ new Location(node.dc, node.rack),
+ NodeVersion.CURRENT)
+ .build().metadata);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org