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

Defer the creation of system_cluster_metadata keyspace until CMS initialization 

* Insert the actual PreInitialize entry in the distribute metadata log
  table using a callback after the log is bootstrapped.
* Remove the implicit insert on subsequent commit, i.e. of the Initialize entry
* Enables the full specifics of PreInit to be encoded in the serialized form,
  removing the special casing for the first CMS member and other nodes
* Correctly invalidate cached KeyspaceMetadata when in a pre-initialized state
* Update Host ids in the system.peers_v2 table directly following CMS initialization
* Initialise gossip/local host id after CMS initialization completes
* Clean up CMS initialization errors which occur after the PreInitialize stage

Co-authored-by: Alex Petrov <oleksandr.petrov@gmail.com>
Co-authored-by: Marcus Eriksson <marcuse@apache.org>
This commit is contained in:
Sam Tunnicliffe 2026-06-29 16:54:02 +01:00
parent 002190748b
commit 896d1d6415
33 changed files with 767 additions and 163 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
6.0-alpha2
* Defer creation of the system_cluster_metadata keyspace until CMS initialization (CASSANDRA-21477)
* Support direct I/O for background SSTable writes (CASSANDRA-21134)
* Relax assertion on partitioner instances in SinglePartitionReadCommand (CASSANDRA-21251)
* Report cancelled read command execution to coordinator as a RequestFailure.TIMEOUT (CASSANDRA-21468)

3
src/java/org/apache/cassandra/db/SystemPeersValidator.java
View File

@ -34,6 +34,7 @@ import org.apache.cassandra.db.virtual.PeersTable;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.membership.Location;
import org.apache.cassandra.tcm.membership.NodeAddresses;
import org.apache.cassandra.tcm.membership.NodeId;
@ -68,6 +69,8 @@ public final class SystemPeersValidator
public static void validateAndRepair(ClusterMetadata metadata)
{
if (metadata.epoch.isBefore(Epoch.FIRST))
return;
Map<InetAddressAndPort, UntypedResultSet.Row> peersV2Rows = getPeersV2Rows();
Map<InetAddress, UntypedResultSet.Row> peersRows = getPeersRows();

31
src/java/org/apache/cassandra/schema/DistributedMetadataLogKeyspace.java
View File

@ -46,6 +46,7 @@ import org.apache.cassandra.tcm.log.LogState;
import org.apache.cassandra.tcm.transformations.cms.PreInitialize;
import org.apache.cassandra.utils.JVMStabilityInspector;
import static org.apache.cassandra.schema.SchemaConstants.METADATA_KEYSPACE_NAME;
import static org.apache.cassandra.tcm.Epoch.FIRST;
public final class DistributedMetadataLogKeyspace
@ -65,7 +66,7 @@ public final class DistributedMetadataLogKeyspace
*/
public static final long GENERATION = 0;
public static final TableId LOG_TABLE_ID = TableId.unsafeDeterministic(SchemaConstants.METADATA_KEYSPACE_NAME, TABLE_NAME);
public static final TableId LOG_TABLE_ID = TableId.unsafeDeterministic(METADATA_KEYSPACE_NAME, TABLE_NAME);
public static final String LOG_TABLE_CQL = "CREATE TABLE %s.%s ("
+ "epoch bigint,"
+ "entry_id bigint,"
@ -80,16 +81,16 @@ public final class DistributedMetadataLogKeyspace
"compaction_window_size","1")))
.build();
public static boolean initialize() throws IOException
public static boolean insertPreInitialize(PreInitialize preInit) throws IOException
{
try
{
String init = String.format("INSERT INTO %s.%s (epoch, transformation, kind, entry_id) " +
"VALUES(?, ?, ?, ?) " +
"IF NOT EXISTS", SchemaConstants.METADATA_KEYSPACE_NAME, TABLE_NAME);
"IF NOT EXISTS", METADATA_KEYSPACE_NAME, TABLE_NAME);
UntypedResultSet result = QueryProcessor.execute(init, ConsistencyLevel.QUORUM,
FIRST.getEpoch(),
Transformation.Kind.PRE_INITIALIZE_CMS.toVersionedBytes(PreInitialize.blank()),
Transformation.Kind.PRE_INITIALIZE_CMS.toVersionedBytes(preInit),
Transformation.Kind.PRE_INITIALIZE_CMS.id,
Entry.Id.NONE.entryId);
@ -124,21 +125,25 @@ public final class DistributedMetadataLogKeyspace
{
try
{
if (previousEpoch.is(FIRST) && !initialize())
// log is not initialized yet this is unexpected
if (previousEpoch.isBefore(FIRST))
{
logger.warn("Previous epoch {} indicates that the {} has not been initialized yet, " +
"not committing entry {}/{} at epoch {}",
previousEpoch, METADATA_KEYSPACE_NAME, entryId, transform, nextEpoch);
return false;
}
// TODO get lowest supported metadata version from ClusterMetadata
ByteBuffer serializedEvent = transform.kind().toVersionedBytes(transform);
ByteBuffer serializedTransform = transform.kind().toVersionedBytes(transform);
String query = String.format("INSERT INTO %s.%s (epoch, entry_id, transformation, kind) " +
"VALUES (?, ?, ?, ?) " +
"IF NOT EXISTS;",
SchemaConstants.METADATA_KEYSPACE_NAME, TABLE_NAME);
METADATA_KEYSPACE_NAME, TABLE_NAME);
UntypedResultSet result = QueryProcessor.execute(query,
ConsistencyLevel.QUORUM,
nextEpoch.getEpoch(),
entryId.entryId,
serializedEvent,
serializedTransform,
transform.kind().id);
return result.one().getBoolean("[applied]");
@ -186,7 +191,7 @@ public final class DistributedMetadataLogKeyspace
// note that we want all entries with epoch >= since - but since we use a reverse partitioner, we actually
// want all entries where the token is less than token(since)
UntypedResultSet resultSet = execute(String.format("SELECT epoch, kind, transformation, entry_id FROM %s.%s WHERE token(epoch) <= token(?)",
SchemaConstants.METADATA_KEYSPACE_NAME, TABLE_NAME),
METADATA_KEYSPACE_NAME, TABLE_NAME),
consistencyLevel, since.getEpoch());
EntryHolder entryHolder = new EntryHolder(since);
for (UntypedResultSet.Row row : resultSet)
@ -237,7 +242,7 @@ public final class DistributedMetadataLogKeyspace
private static TableMetadata.Builder parse(String cql, String table, String description)
{
return CreateTableStatement.parse(String.format(cql, SchemaConstants.METADATA_KEYSPACE_NAME, table), SchemaConstants.METADATA_KEYSPACE_NAME)
return CreateTableStatement.parse(String.format(cql, METADATA_KEYSPACE_NAME, table), METADATA_KEYSPACE_NAME)
.id(LOG_TABLE_ID)
.epoch(FIRST)
.comment(description);
@ -245,7 +250,7 @@ public final class DistributedMetadataLogKeyspace
public static KeyspaceMetadata initialMetadata(Set<String> knownDatacenters)
{
return KeyspaceMetadata.create(SchemaConstants.METADATA_KEYSPACE_NAME, new KeyspaceParams(true, ReplicationParams.simpleMeta(1, knownDatacenters), FastPathStrategy.simple()), Tables.of(Log));
return KeyspaceMetadata.create(METADATA_KEYSPACE_NAME, new KeyspaceParams(true, ReplicationParams.simpleMeta(1, knownDatacenters), FastPathStrategy.simple()), Tables.of(Log));
}
public static KeyspaceMetadata initialMetadata(String datacenter)

63
src/java/org/apache/cassandra/schema/DistributedSchema.java
View File

@ -21,14 +21,13 @@ package org.apache.cassandra.schema;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.stream.Collectors;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
@ -65,25 +64,6 @@ public class DistributedSchema implements MetadataValue<DistributedSchema>
return new DistributedSchema(Keyspaces.none(), Epoch.EMPTY);
}
public static DistributedSchema first(Set<String> knownDatacenters)
{
// During upgrades from pre-6.0 versions, the replication params of the system_cluster_metadata
// keyspace using one of the existing DCs. This is so that this keyspace does not cause issues
// for tooling, clients or control plane systems which may inspect schema and have specific
// expectations about DC layout. This keyspace is unused until the CMS is initialized.
// For new clusters which start out on 6.0 or later, this is not necessary to the initial
// replication params use a empty string for the placeholder DC name.
// During CMS initialization, the replication of this keyspace will be set for real using
// the DC of the first node to become a CMS member. This happens in the PreInitialize
// transformation when executed on the first CMS member.
if (knownDatacenters.isEmpty())
knownDatacenters = Collections.singleton("");
return new DistributedSchema(Keyspaces.of(DistributedMetadataLogKeyspace.initialMetadata(knownDatacenters)), Epoch.FIRST);
}
private static ImmutableMap<TableId, TableMetadata> keyspacesToTableMap(Keyspaces keyspaces)
{
ImmutableMap.Builder<TableId, TableMetadata> builder = ImmutableMap.builder();
@ -159,28 +139,23 @@ public class DistributedSchema implements MetadataValue<DistributedSchema>
* @deprecated since TCM, used on upgrade from gossip to populate system schema tables with the correct generation
*/
@Deprecated(since = "TCM")
public static List<Pair<KeyspaceMetadata, Long>> distributedKeyspacesWithGeneration(Set<String> knownDatacenters)
public static List<Pair<KeyspaceMetadata, Long>> distributedKeyspacesWithGeneration()
{
return ImmutableList.of(Pair.create(DistributedMetadataLogKeyspace.initialMetadata(knownDatacenters), DistributedMetadataLogKeyspace.GENERATION),
Pair.create(TraceKeyspace.metadata(), TraceKeyspace.GENERATION),
return ImmutableList.of(Pair.create(TraceKeyspace.metadata(), TraceKeyspace.GENERATION),
Pair.create(SystemDistributedKeyspace.metadata(), SystemDistributedKeyspace.GENERATION),
Pair.create(AuthKeyspace.metadata(),AuthKeyspace.GENERATION));
}
public static DistributedSchema fromSystemTables(Keyspaces keyspaces, Set<String> knownDatacenters)
public static DistributedSchema fromSystemTables(Keyspaces keyspaces)
{
if (!keyspaces.containsKeyspace(SchemaConstants.METADATA_KEYSPACE_NAME))
{
Keyspaces kss = Keyspaces.none();
for (Pair<KeyspaceMetadata, Long> ksmGen : distributedKeyspacesWithGeneration(knownDatacenters))
kss = kss.with(ksmGen.left);
for (KeyspaceMetadata ksm : keyspaces) // on disk keyspaces
kss = kss.withAddedOrUpdated(kss.get(ksm.name)
.map(k -> merged(ksm, k))
.orElse(ksm));
keyspaces = kss;
}
return new DistributedSchema(keyspaces, Epoch.UPGRADE_GOSSIP);
Keyspaces kss = Keyspaces.none();
for (Pair<KeyspaceMetadata, Long> ksmGen : distributedKeyspacesWithGeneration())
kss = kss.with(ksmGen.left);
for (KeyspaceMetadata ksm : keyspaces) // on disk keyspaces
kss = kss.withAddedOrUpdated(kss.get(ksm.name)
.map(k -> merged(ksm, k))
.orElse(ksm));
return new DistributedSchema(kss, Epoch.UPGRADE_GOSSIP);
}
/**
@ -268,7 +243,8 @@ public class DistributedSchema implements MetadataValue<DistributedSchema>
}
});
// Avoid system table side effects during initialization
// Avoid system table side effects before initialization, otherwise mismatching schema can block CMS
// progress as nodes report disagreement. Also, tooling should not mutate system tables.
if (epoch.isEqualOrAfter(Epoch.FIRST) && !DatabaseDescriptor.isClientOrToolInitialized())
{
Collection<Mutation> mutations = SchemaKeyspace.convertSchemaDiffToMutations(ksDiff, FBUtilities.timestampMicros());
@ -481,6 +457,17 @@ public class DistributedSchema implements MetadataValue<DistributedSchema>
});
}
public String conciseToString()
{
return keyspaces.stream()
.map(ksm -> ksm.name + ": {" +
ksm.tables.stream()
.map(t -> t.name)
.collect(Collectors.joining(","))
+ '}')
.collect(Collectors.joining(","));
}
public static class Serializer implements MetadataSerializer<DistributedSchema>
{
public void serialize(DistributedSchema t, DataOutputPlus out, Version version) throws IOException

14
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -2151,9 +2151,17 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
else
{
// Handling the keyspaces which are not handled by CMS like system keyspace which uses LocalStrategy.
AbstractReplicationStrategy strategy = Keyspace.open(keyspace).getReplicationStrategy();
for (Range<Token> range : ranges)
rangeToEndpointMap.put(range, strategy.calculateNaturalReplicas(range.right, metadata));
Keyspace ks = Keyspace.openIfExists(keyspace);
if (ks != null)
{
AbstractReplicationStrategy strategy = ks.getReplicationStrategy();
for (Range<Token> range : ranges)
rangeToEndpointMap.put(range, strategy.calculateNaturalReplicas(range.right, metadata));
}
else
{
throw new IllegalArgumentException("Unknown keyspace " + keyspace);
}
}
return new EndpointsByRange(rangeToEndpointMap);

2
src/java/org/apache/cassandra/tcm/AbstractLocalProcessor.java
View File

@ -196,4 +196,4 @@ public abstract class AbstractLocalProcessor implements Processor
public abstract ClusterMetadata fetchLogAndWait(Epoch waitFor, Retry retryPolicy);
protected abstract boolean tryCommitOne(Entry.Id entryId, Transformation transform, Epoch previousEpoch, Epoch nextEpoch);
}
}

4
src/java/org/apache/cassandra/tcm/CMSOperations.java
View File

@ -154,14 +154,14 @@ public class CMSOperations implements CMSOperationsMBean
ClusterMetadata metadata = ClusterMetadata.current();
String members = metadata.fullCMSMembers().stream().sorted().map(Object::toString).collect(Collectors.joining(","));
info.put(MEMBERS, members);
info.put(NEEDS_RECONFIGURATION, Boolean.toString(needsReconfiguration(metadata)));
info.put(NEEDS_RECONFIGURATION, Boolean.toString(metadata.epoch.isBefore(Epoch.FIRST) || needsReconfiguration(metadata)));
info.put(IS_MEMBER, Boolean.toString(cms.isCurrentMember(FBUtilities.getBroadcastAddressAndPort())));
info.put(SERVICE_STATE, ClusterMetadataService.state(metadata).toString());
info.put(IS_MIGRATING, Boolean.toString(cms.isMigrating()));
info.put(EPOCH, Long.toString(metadata.epoch.getEpoch()));
info.put(LOCAL_PENDING, Integer.toString(cms.log().pendingBufferSize()));
info.put(COMMITS_PAUSED, Boolean.toString(cms.commitsPaused()));
info.put(REPLICATION_FACTOR, ReplicationParams.meta(metadata).toString());
info.put(REPLICATION_FACTOR, metadata.epoch.isBefore(Epoch.FIRST) ? "" : ReplicationParams.meta(metadata).toString());
info.put(CMS_ID, Integer.toString(metadata.metadataIdentifier));
return info;
}

23
src/java/org/apache/cassandra/tcm/ClusterMetadata.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.tcm;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@ -131,7 +132,7 @@ public class ClusterMetadata
@VisibleForTesting
public ClusterMetadata(IPartitioner partitioner, Directory directory)
{
this(partitioner, directory, DistributedSchema.first(directory.knownDatacenters()));
this(partitioner, directory, DistributedSchema.empty());
}
@VisibleForTesting
@ -216,6 +217,9 @@ public class ClusterMetadata
public Set<InetAddressAndPort> fullCMSMembers()
{
if (epoch.isBefore(Epoch.FIRST))
return Collections.emptySet();
if (fullCMSEndpoints == null)
this.fullCMSEndpoints = ImmutableSet.copyOf(placements.get(ReplicationParams.meta(this)).reads.byEndpoint().keySet());
return fullCMSEndpoints;
@ -223,6 +227,9 @@ public class ClusterMetadata
public Set<NodeId> fullCMSMemberIds()
{
if (epoch.isBefore(Epoch.FIRST))
return Collections.emptySet();
if (fullCMSIds == null)
this.fullCMSIds = placements.get(ReplicationParams.meta(this)).reads.byEndpoint().keySet().stream().map(directory::peerId).collect(toImmutableSet());
return fullCMSIds;
@ -230,6 +237,9 @@ public class ClusterMetadata
public EndpointsForRange fullCMSMembersAsReplicas()
{
if (epoch.isBefore(Epoch.FIRST))
return EndpointsForRange.empty(MetaStrategy.entireRange);
if (fullCMSReplicas == null)
fullCMSReplicas = placements.get(ReplicationParams.meta(this)).reads.forRange(MetaStrategy.entireRange).get();
return fullCMSReplicas;
@ -309,7 +319,7 @@ public class ClusterMetadata
@SuppressWarnings("unchecked")
private static <V> V capLastModified(MetadataValue<V> value, Epoch maxEpoch)
{
return value == null || value.lastModified().isEqualOrBefore(maxEpoch)
return value == null || (value.lastModified().isEqualOrAfter(Epoch.EMPTY) && value.lastModified().isEqualOrBefore(maxEpoch))
? (V)value
: value.withLastModified(maxEpoch);
}
@ -946,6 +956,15 @@ public class ClusterMetadata
'}';
}
public String conciseToString()
{
return "ClusterMetadata{" + "epoch=" + epoch +
", schema=" + schema.conciseToString() +
", directory=" + directory.conciseToString(tokenMap.asMap()) +
", placements=" + placements.conciseToString() +
'}';
}
@Override
public boolean equals(Object o)
{

64
src/java/org/apache/cassandra/tcm/ClusterMetadataService.java
View File

@ -29,6 +29,7 @@ import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
@ -52,6 +53,7 @@ import org.apache.cassandra.metrics.TCMMetrics;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessageDelivery;
import org.apache.cassandra.schema.DistributedMetadataLogKeyspace;
import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.schema.Keyspaces;
import org.apache.cassandra.schema.ReplicationParams;
@ -79,6 +81,7 @@ import org.apache.cassandra.tcm.serialization.VerboseMetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.tcm.transformations.ForceSnapshot;
import org.apache.cassandra.tcm.transformations.TriggerSnapshot;
import org.apache.cassandra.tcm.transformations.cms.PreInitialize;
import org.apache.cassandra.tcm.transformations.cms.PrepareCMSReconfiguration;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JVMStabilityInspector;
@ -138,6 +141,14 @@ public class ClusterMetadataService
private final LocalLog log;
private final MetadataSnapshots snapshots;
/*
* Special callback for execution when the PreInitialize transformation is used to bootstrap the cluster metadata
* log. In practice, this is only relevant when using PaxosBackedProcessor which needs to insert the PreInitialize
* entry into the distributed log table, but is unable to do so until it has been enacted, the keyspace created and
* the replication configured.
*/
private final Consumer<PreInitialize> logBootstrapCallback;
private final IVerbHandler<LogState> replicationHandler;
private final IVerbHandler<LogState> logNotifyHandler;
private final IVerbHandler<FetchCMSLog> fetchLogHandler;
@ -181,13 +192,17 @@ public class ClusterMetadataService
if (CassandraRelevantProperties.TCM_USE_ATOMIC_LONG_PROCESSOR.getBoolean())
{
log = logSpec.sync().withStorage(new AtomicLongBackedProcessor.InMemoryStorage()).createLog();
localProcessor = wrapProcessor.apply(new AtomicLongBackedProcessor(log, logSpec.isReset()));
AtomicLongBackedProcessor processor = new AtomicLongBackedProcessor(log, logSpec.isReset());
logBootstrapCallback = logBootstrapCallback(processor);
localProcessor = wrapProcessor.apply(processor);
fetchLogHandler = new FetchCMSLog.Handler((e, ignored) -> logSpec.storage().getLogState(e));
}
else
{
log = logSpec.async().createLog();
localProcessor = wrapProcessor.apply(new PaxosBackedProcessor(log));
PaxosBackedProcessor processor = new PaxosBackedProcessor(log);
logBootstrapCallback = logBootstrapCallback(processor);
localProcessor = wrapProcessor.apply(processor);
fetchLogHandler = new FetchCMSLog.Handler();
}
@ -234,8 +249,12 @@ public class ClusterMetadataService
commitRequestHandler = isMemberOfOwnershipGroup ? new Commit.Handler(processor, replicator, () -> LOCAL) : null;
peerLogFetcher = new PeerLogFetcher(log);
logBootstrapCallback = logBootstrapCallback(processor);
}
/**
* Only called from initializeForTools
*/
private ClusterMetadataService(PlacementProvider placementProvider,
MetadataSnapshots snapshots,
LocalLog log,
@ -257,6 +276,7 @@ public class ClusterMetadataService
this.fetchLogHandler = fetchLogHandler;
this.commitRequestHandler = commitRequestHandler;
this.peerLogFetcher = peerLogFetcher;
this.logBootstrapCallback = logBootstrapCallback(processor);
}
@SuppressWarnings("resource")
@ -265,8 +285,7 @@ public class ClusterMetadataService
if (instance != null)
return;
String localDC = DatabaseDescriptor.getLocalDataCenter();
ClusterMetadata emptyFromSystemTables = emptyWithSchemaFromSystemTables(Collections.singleton(localDC))
.forceEpoch(Epoch.EMPTY);
ClusterMetadata emptyFromSystemTables = emptyWithSchemaFromSystemTables().forceEpoch(Epoch.EMPTY);
LocalLog.LogSpec logSpec = LocalLog.logSpec()
.withInitialState(emptyFromSystemTables)
@ -294,7 +313,7 @@ public class ClusterMetadataService
new PeerLogFetcher(log));
log.readyUnchecked();
log.bootstrap(FBUtilities.getBroadcastAddressAndPort(), localDC);
log.bootstrap(FBUtilities.getBroadcastAddressAndPort(), localDC, (p) -> {});
ClusterMetadataService.setInstance(cms);
}
@ -336,10 +355,43 @@ public class ClusterMetadataService
null);
log.readyUnchecked();
log.bootstrap(FBUtilities.getBroadcastAddressAndPort(), localDC);
log.bootstrap(FBUtilities.getBroadcastAddressAndPort(), localDC, (p) -> {});
ClusterMetadataService.setInstance(cms);
}
/*
* Hook to be executed when the LocalLog is bootstrapped with the PreInitialize transformation. This is done on
* the first CMS member to set up the initial replication and data placements for the metadata keyspace.
*/
public Consumer<PreInitialize> logBootstrapCallback()
{
return logBootstrapCallback;
}
private static Consumer<PreInitialize> logBootstrapCallback(Processor processor)
{
if (processor instanceof PaxosBackedProcessor)
{
// Insert an entry containing the PRE_INITIALIZE_CMS transform at Epoch.FIRST in the distributed
// log table. This can only be done after the log is bootstrapped as it depends on the effects of
// that transform on ClusterMetadata.
return preInit -> {
try
{
if (DistributedMetadataLogKeyspace.insertPreInitialize(preInit))
logger.info("Successfully inserted pre-initialize entry into distributed metadata log");
else
throw new IllegalStateException("Failed to insert pre-initialize entry into distributed metadata log. Check server for details");
}
catch (IOException e)
{
throw new IllegalStateException("Unable to pre-initialize distributed metadata log table", e);
}
};
}
// otherwise, this is a noop.
return preInit -> {};
}
@SuppressWarnings("resource")
public static void initializeForClients()

33
src/java/org/apache/cassandra/tcm/Startup.java
View File

@ -65,6 +65,7 @@ import org.apache.cassandra.tcm.log.LogStorage;
import org.apache.cassandra.tcm.log.SystemKeyspaceStorage;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.membership.NodeState;
import org.apache.cassandra.tcm.migration.CMSInitializationException;
import org.apache.cassandra.tcm.migration.CMSInitializationRequest;
import org.apache.cassandra.tcm.migration.Election;
import org.apache.cassandra.tcm.ownership.UniformRangePlacement;
@ -134,11 +135,14 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
}
/**
* Make this node a _first_ CMS node.
* Make this node the _first_ CMS node.
* <p>
* (1) Append PreInitialize transformation to local in-memory log. When distributed metadata keyspace is initialized, a no-op transformation will
* be added to other nodes. This is required since as of now, no node actually owns distributed metadata keyspace.
* (2) Commit Initialize transformation, which holds a snapshot of metadata as of now.
* (1) Append PreInitialize transformation to local in-memory log.
* (1a) Once this enacted and the distributed metadata keyspace is initialized, the PreInitialize transformation
* will be inserted into the log table. This is required since as before this point, the keyspace was not availble
* or configured with any replication or placements.
* (2) Commit Initialize transformation, which holds a complete snapshot of metadata as of now.
* Other nodes in the cluster, if there are any, will receive both of these log entries and enact them locally.
* <p>
* This process is applicable for gossip upgrades as well as regular vote-and-startup process.
*/
@ -146,11 +150,17 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
{
InetAddressAndPort addr = FBUtilities.getBroadcastAddressAndPort();
String datacenter = DatabaseDescriptor.getLocator().local().datacenter;
ClusterMetadataService.instance().log().bootstrap(addr, datacenter);
ClusterMetadataService cms = ClusterMetadataService.instance();
cms.log().bootstrap(addr, datacenter, cms.logBootstrapCallback());
ClusterMetadata metadata = ClusterMetadata.current();
assert ClusterMetadataService.state() == LOCAL : String.format("Can't initialize as node hasn't transitioned to CMS state. State: %s.\n%s", ClusterMetadataService.state(), metadata);
Initialize initialize = new Initialize(metadata.initializeClusterIdentifier(addr.hashCode()));
ClusterMetadataService.instance().commit(initialize);
ClusterMetadataService.instance().commit(initialize,
m -> { logger.info("INITIALIZE_CMS committed successfully"); return m;},
(code, message) -> {
logger.info("INITIALIZE_CMS commit failure: ({}) {}", code, message);
throw new CMSInitializationException();
});
}
public static void initializeAsNonCmsNode(Function<Processor, Processor> wrapProcessor) throws StartupException
@ -262,9 +272,9 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
Election.instance.migrated();
}
private static void updateSystemSchemaTables(Set<String> knownDatacenters)
private static void updateSystemSchemaTables()
{
List<Pair<KeyspaceMetadata, Long>> kss = DistributedSchema.distributedKeyspacesWithGeneration(knownDatacenters);
List<Pair<KeyspaceMetadata, Long>> kss = DistributedSchema.distributedKeyspacesWithGeneration();
List<Mutation> mutations = new ArrayList<>();
for (Pair<KeyspaceMetadata, Long> ksm : kss)
{
@ -279,9 +289,8 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
*/
public static void initializeFromGossip(Function<Processor, Processor> wrapProcessor, Runnable initMessaging) throws StartupException
{
Set<String> knownDcs = SystemKeyspace.allKnownDatacenters();
updateSystemSchemaTables(knownDcs);
ClusterMetadata emptyFromSystemTables = emptyWithSchemaFromSystemTables(knownDcs);
updateSystemSchemaTables();
ClusterMetadata emptyFromSystemTables = emptyWithSchemaFromSystemTables();
LocalLog.LogSpec logSpec = LocalLog.logSpec()
.withInitialState(emptyFromSystemTables)
.afterReplay(Startup::scrubDataDirectories,
@ -332,7 +341,7 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
logger.debug("Got epStates {}", epStates);
ClusterMetadata initial = fromEndpointStates(emptyFromSystemTables.schema, epStates);
logger.debug("Created initial ClusterMetadata {}", initial);
logger.info("Created initial ClusterMetadata {}", initial.conciseToString());
ClusterMetadataService.instance().setFromGossip(initial);
Gossiper.instance.clearUnsafe();

4
src/java/org/apache/cassandra/tcm/compatibility/GossipHelper.java
View File

@ -287,7 +287,7 @@ public class GossipHelper
return NodeVersion.fromCassandraVersion(cassandraVersion);
}
public static ClusterMetadata emptyWithSchemaFromSystemTables(Set<String> allKnownDatacenters)
public static ClusterMetadata emptyWithSchemaFromSystemTables()
{
// If this instance was previously upgraded then subsequently downgraded, the metadata keyspace may have been
// added to system_schema tables. If so, don't include it in the initial schema as this will cause it to be
@ -297,7 +297,7 @@ public class GossipHelper
.filter(k -> !k.name.equals(SchemaConstants.METADATA_KEYSPACE_NAME));
return new ClusterMetadata(Epoch.UPGRADE_STARTUP,
DatabaseDescriptor.getPartitioner(),
DistributedSchema.fromSystemTables(keyspaces, allKnownDatacenters),
DistributedSchema.fromSystemTables(keyspaces),
Directory.EMPTY,
new TokenMap(DatabaseDescriptor.getPartitioner()),
DataPlacements.empty(),

7
src/java/org/apache/cassandra/tcm/listeners/LegacyStateListener.java
View File

@ -41,6 +41,7 @@ import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.MultiStepOperation;
import org.apache.cassandra.tcm.compatibility.GossipHelper;
import org.apache.cassandra.tcm.membership.Directory;
@ -73,8 +74,12 @@ public class LegacyStateListener implements ChangeListener
Set<NodeId> changed = new HashSet<>();
for (NodeId node : next.directory.peerIds())
{
if (directoryEntryChangedFor(node, prev.directory, next.directory) || !prev.tokenMap.tokens(node).equals(next.tokenMap.tokens(node)))
if (prev.epoch.isEqualOrBefore(Epoch.FIRST)
|| directoryEntryChangedFor(node, prev.directory, next.directory)
|| !prev.tokenMap.tokens(node).equals(next.tokenMap.tokens(node)))
{
changed.add(node);
}
}
// next.myNodeId() can be null during replay (before we have registered) but if it is present and

17
src/java/org/apache/cassandra/tcm/listeners/UpgradeMigrationListener.java
View File

@ -21,9 +21,11 @@ package org.apache.cassandra.tcm.listeners;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.utils.CassandraVersion;
/**
@ -42,11 +44,20 @@ public class UpgradeMigrationListener implements ChangeListener
{
if (prev.epoch.equals(Epoch.UPGRADE_GOSSIP))
{
logger.info("Detected upgrade from gossip mode, updating my host id in gossip to {}", next.myNodeId());
Gossiper.instance.mergeNodeToGossip(next.myNodeId(), next);
logger.info("Detected upgrade from gossip mode");
return;
}
else if (prev.epoch.equals(Epoch.FIRST) && !next.directory.isEmpty()) // directory is non-empty after initialization during gossip upgrade
{
NodeId localId = next.myNodeId();
if (localId != null)
{
logger.info("Initialized CMS, updating local host id to {}", next.myNodeId());
SystemKeyspace.setLocalHostId(next.myNodeId().toUUID());
Gossiper.instance.mergeNodeToGossip(next.myNodeId(), next);
}
if (Gossiper.instance.getQuarantineDisabled())
Gossiper.instance.clearQuarantinedEndpoints();
return;
}
CassandraVersion prevMinVersion = prev.directory.clusterMinVersion.cassandraVersion;

32
src/java/org/apache/cassandra/tcm/log/LocalLog.java
View File

@ -32,6 +32,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.stream.Collectors;
@ -277,22 +278,24 @@ public abstract class LocalLog implements Closeable
@VisibleForTesting
public void unsafeBootstrapForTesting(InetAddressAndPort addr)
{
bootstrap(addr, "");
bootstrap(addr, "", (p) -> {});
}
/**
*
* @param addr
* @param datacenter
* Set up the initial state of the local ClusterMetadata, enacting the PreInitialize transform which sets up the
* distributed metadata log keyspace, its replication, and data placements.
* @param addr Address of the first CMS member, this should be the local node address
* @param datacenter DC of the first CMS member
*/
public void bootstrap(InetAddressAndPort addr, String datacenter)
public void bootstrap(InetAddressAndPort addr, String datacenter, Consumer<PreInitialize> postBootstrap)
{
ClusterMetadata metadata = metadata();
assert metadata.epoch.isBefore(FIRST) : String.format("Metadata epoch %s should be before first", metadata.epoch);
Transformation transform = PreInitialize.withFirstCMS(addr, datacenter);
PreInitialize transform = PreInitialize.withFirstCMS(addr, datacenter);
append(new Entry(Entry.Id.NONE, FIRST, transform));
metadata = waitForHighestConsecutive();
assert metadata.epoch.is(Epoch.FIRST) : String.format("Epoch: %s. CMS: %s", metadata.epoch, metadata.fullCMSMembers());
postBootstrap.accept(transform);
}
public LogStorage storage()
@ -475,8 +478,15 @@ public abstract class LocalLog implements Closeable
ClusterMetadata prev = committed.get();
// ForceSnapshot + Bootstrap entries can "jump" epoch
boolean isPreInit = pendingEntry.transform.kind() == Transformation.Kind.PRE_INITIALIZE_CMS;
boolean isSnapshot = pendingEntry.transform.kind() == Transformation.Kind.FORCE_SNAPSHOT;
Transformation.Kind kind = pendingEntry.transform.kind();
boolean isPreInit = kind == Transformation.Kind.PRE_INITIALIZE_CMS;
boolean isInit = kind == Transformation.Kind.INITIALIZE_CMS;
boolean isSnapshot = kind == Transformation.Kind.FORCE_SNAPSHOT ;
// only a PRE_INITIALIZE_CMS or a snapshot is allowed to skip over gaps.
// Note: INITIALIZE_CMS is not allowed to do this as during upgrades it can allow us to jump over the
// PRE_INITIALIZE_CMS entry if the INITIALIZE_CMS is received first. This then creates a gap at Epoch.FIRST
// which can never be resolved. In turn, that makes it impossible to build a LogState for replay purposes
// with a correct and consecutive set of entries if the node is bounced before applying a later snapshot.
if (pendingEntry.epoch.isDirectlyAfter(prev.epoch)
|| ((isPreInit || isSnapshot) && pendingEntry.epoch.isAfter(prev.epoch)))
{
@ -512,7 +522,7 @@ public abstract class LocalLog implements Closeable
if (replayComplete.get() && pendingEntry.transform.kind() != Transformation.Kind.FORCE_SNAPSHOT)
storage.append(pendingEntry.maybeUnwrapExecuted());
notifyPreCommit(prev, next, isSnapshot);
notifyPreCommit(prev, next, isSnapshot || isInit);
if (committed.compareAndSet(prev, next))
{
@ -528,7 +538,7 @@ public abstract class LocalLog implements Closeable
next.epoch, prev.epoch, metadata().epoch));
}
notifyPostCommit(prev, next, isSnapshot);
notifyPostCommit(prev, next, isSnapshot || isInit);
}
catch (StopProcessingException t)
{
@ -917,7 +927,7 @@ public abstract class LocalLog implements Closeable
private LogListener snapshotListener()
{
return (entry, metadata) -> {
if (ClusterMetadataService.state() != ClusterMetadataService.State.LOCAL)
if (entry.epoch.isEqualOrBefore(Epoch.FIRST) || ClusterMetadataService.state() != ClusterMetadataService.State.LOCAL)
return;
if ((entry.epoch.getEpoch() % DatabaseDescriptor.getMetadataSnapshotFrequency()) == 0)

26
src/java/org/apache/cassandra/tcm/membership/Directory.java
View File

@ -42,6 +42,7 @@ import accord.utils.Invariants;
import accord.utils.btree.BTreeSet;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.locator.InetAddressAndPort;
@ -50,6 +51,7 @@ import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.MetadataValue;
import org.apache.cassandra.tcm.serialization.MetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.utils.SortedBiMultiValMap;
import org.apache.cassandra.utils.UUIDSerializer;
import org.apache.cassandra.utils.btree.BTreeBiMap;
import org.apache.cassandra.utils.btree.BTreeMap;
@ -524,6 +526,30 @@ public class Directory implements MetadataValue<Directory>
.collect(Collectors.joining("\n"));
}
/**
* nodeId(address): {state, tokens, ser ver, cass ver, dc:rack}
*/
public String conciseToString(SortedBiMultiValMap<Token, NodeId> tokenMap)
{
return peers.entrySet().stream().map((peerEntry) -> {
StringBuilder sb = new StringBuilder();
NodeId nodeId = peerEntry.getKey();
sb.append(nodeId.id()).append('(').append(peerEntry.getValue()).append("): ");
sb.append('{');
sb.append(states.get(nodeId).ordinal());
if (tokenMap != null)
sb.append(',').append(tokenMap.inverse().get(nodeId));
NodeVersion version = versions.get(nodeId);
if (version != null)
sb.append(',').append(version.serializationVersion).append(',').append(version.cassandraVersion);
Location loc = locations.get(nodeId);
if (loc != null)
sb.append(',').append(loc.datacenter).append(':').append(loc.rack);
sb.append('}');
return sb.toString();
}).collect(Collectors.joining(","));
}
private static class Node
{
public static final Serializer serializer = new Serializer();

27
src/java/org/apache/cassandra/tcm/migration/CMSInitializationException.java Normal file
View File

@ -0,0 +1,27 @@
/*
* 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.migration;
public class CMSInitializationException extends RuntimeException
{
public CMSInitializationException()
{
super("CMS initialization failed, see logs for details");
}
}

10
src/java/org/apache/cassandra/tcm/migration/CMSInitializationRequest.java
View File

@ -68,6 +68,16 @@ public class CMSInitializationRequest
this.schemaVersion = schemaVersion;
}
@Override
public String toString()
{
return "CMSInitializationRequest{" +
"initiator=" + initiator +
", directory=" + directory.conciseToString(tokenMap.asMap()) +
", schemaVersion=" + schemaVersion +
'}';
}
public static class Serializer implements IVersionedSerializer<CMSInitializationRequest>
{
private final Version serializationVersion;

20
src/java/org/apache/cassandra/tcm/migration/Election.java
View File

@ -33,6 +33,8 @@ import com.google.common.collect.Sets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.IVerbHandler;
@ -41,8 +43,11 @@ import org.apache.cassandra.net.MessageDelivery;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.schema.DistributedMetadataLogKeyspace;
import org.apache.cassandra.schema.Keyspaces;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.SchemaKeyspace;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.Startup;
import org.apache.cassandra.tcm.membership.Directory;
@ -87,6 +92,8 @@ public class Election
public void nominateSelf(Set<InetAddressAndPort> candidates, Set<InetAddressAndPort> ignoredEndpoints, ClusterMetadata metadata, boolean verifyAllPeersMetadata)
{
// Note: this is probably identical to the supplied metadata, but not guaranteed to be
ClusterMetadata priorState = ClusterMetadata.current();
Set<InetAddressAndPort> sendTo = new HashSet<>(candidates);
sendTo.removeAll(ignoredEndpoints);
sendTo.remove(FBUtilities.getBroadcastAddressAndPort());
@ -102,6 +109,19 @@ public class Election
{
logger.error("Got error nominating self", e);
abort(initializationRequest.initiator, sendTo);
ClusterMetadata currentState = ClusterMetadata.current();
// Clean up the system_cluster_metadata keyspace which may have been created by PRE_INITIALIZE_CMS
Keyspace metaKeyspace = currentState.schema.getKeyspace(SchemaConstants.METADATA_KEYSPACE_NAME);
if (metaKeyspace != null)
metaKeyspace.unload(true);
// Remove entries from system_schema tables
Keyspaces.KeyspacesDiff diff = Keyspaces.diff(currentState.schema.getKeyspaces(), priorState.schema.getKeyspaces());
Collection<Mutation> mutations = SchemaKeyspace.convertSchemaDiffToMutations(diff, FBUtilities.timestampMicros());
SchemaKeyspace.applyChanges(mutations);
ClusterMetadataService.instance().log().unsafeSetCommittedFromGossip(priorState);
throw e;
}
}

6
src/java/org/apache/cassandra/tcm/ownership/DataPlacements.java
View File

@ -24,6 +24,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
@ -219,6 +220,11 @@ public class DataPlacements extends ReplicationMap<DataPlacement> implements Met
return new Builder(new HashMap<>(this.asMap()));
}
public String conciseToString()
{
return keys().stream().map(ReplicationParams::toString).collect(Collectors.joining(","));
}
public static class Builder
{
private final Map<ReplicationParams, DataPlacement> map;

4
src/java/org/apache/cassandra/tcm/transformations/cms/Initialize.java
View File

@ -23,12 +23,12 @@ import java.io.IOException;
import org.apache.cassandra.auth.AuthKeyspace;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.locator.MetaStrategy;
import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.schema.Keyspaces;
import org.apache.cassandra.schema.SystemDistributedKeyspace;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Transformation;
import org.apache.cassandra.tcm.sequences.LockedRanges;
import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.tcm.transformations.ForceSnapshot;
@ -75,7 +75,7 @@ public class Initialize extends ForceSnapshot
? setUpDistributedSystemKeyspaces(next)
: next.schema.getKeyspaces());
ClusterMetadata.Transformer transformer = next.transformer().with(initialSchema);
return Transformation.success(transformer, MetaStrategy.affectedRanges(prev));
return Transformation.success(transformer, LockedRanges.AffectedRanges.EMPTY);
}
public Keyspaces setUpDistributedSystemKeyspaces(ClusterMetadata next)

41
src/java/org/apache/cassandra/tcm/transformations/cms/PreInitialize.java
View File

@ -52,11 +52,6 @@ public class PreInitialize implements Transformation
this.datacenter = datacenter;
}
public static PreInitialize forTesting()
{
return new PreInitialize(null, null);
}
public static PreInitialize blank()
{
return new PreInitialize(null, null);
@ -78,23 +73,17 @@ public class PreInitialize implements Transformation
ClusterMetadata.Transformer transformer = metadata.transformer();
// This null check is a leftover from previous implementations. In earlier versions the address and datacenter
// were not be included in the serialized form of this transform and so were not written to the local or
// distributed logs nor included in the log entries sent over the wire between peers.
// The null check remains to handle log entries written in that legacy format. The log entry which immediately
// follows PRE_INITIALIZE_CMS must contain an INITIALIZE_CMS transform, which will necessarily include the
// distributed metadata keyspace definition with the replication settings bootstrapped by PRE_INITIALIZE. This
// full ClusterMetadata becomes the starting point upon which further log entries are applied. The ultimate
// effect of this sequence is that once INITIALIZE_CMS has been committed to the log, the actual content of
// PRE_INITIALIZE_CMS becomes irrelevant.
if (addr != null)
{
// If addr != null, then this is being executed on the peer which is actually initializing the log
// for the very first time.
// addr and datacenter are only used to bootstrap the replication of the distributed metatada
// keyspace on the first CMS node. They are never serialized into the distributed metadata log or
// passed to any other peer.
//
// PRE_INITIALIZE_CMS @ Epoch.FIRST, must be followed in the log by INITIALIZE_CMS @ (Epoch.FIRST + 1).
// The serialization of INITIALIZE_CMS includes the full ClusterMetadata at that point, which is
// obviously minimal, but will necessarily include the distributed metadata keyspace definition with
// the replication settings bootstrapped by PRE_INITIALIZE. This full ClusterMetadata becomes the
// starting point upon which further log entries are applied. So this means that once INITIALIZE_CMS
// has been committed to the log, the actual content of PRE_INITIALIZE_CMS is irrelevant, even on
// the first CMS node if it happens to replay it from its local storage after a restart.
DataPlacement.Builder dataPlacementBuilder = DataPlacement.builder();
Replica replica = new Replica(addr,
MetaStrategy.partitioner.getMinimumToken(),
@ -107,11 +96,11 @@ public class PreInitialize implements Transformation
dataPlacementBuilder.build()).build();
transformer.with(initialPlacement);
// re-initialise the schema distributed metadata keyspace so it gets the
// correct replication settings based on the DC of the initial CMS node
// create the distributed metadata keyspace in schema with replication settings based on the DC of the
// initial CMS node
Keyspaces updated = metadata.schema.getKeyspaces()
.withAddedOrReplaced(DistributedMetadataLogKeyspace.initialMetadata(datacenter));
transformer.with(new DistributedSchema(updated));
transformer.with(new DistributedSchema(updated, Epoch.FIRST));
}
ClusterMetadata.Transformer.Transformed transformed = transformer.build();
@ -121,9 +110,13 @@ public class PreInitialize implements Transformation
return new Success(metadata, LockedRanges.AffectedRanges.EMPTY, transformed.modifiedKeys);
}
@Override
public String toString()
{
return "PreInitialize";
return "PreInitialize{" +
"addr=" + addr +
", datacenter='" + datacenter + '\'' +
'}';
}
public static class Serializer implements AsymmetricMetadataSerializer<Transformation, PreInitialize>

12
test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataSingleNodeUpgradeTest.java
View File

@ -26,6 +26,7 @@ import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.service.StorageService;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
/**
* @see org.apache.cassandra.tools.nodetool.CMSAdmin.InitializeCMS
@ -46,8 +47,17 @@ public class ClusterMetadataSingleNodeUpgradeTest extends UpgradeTestBase
cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
})
.runAfterClusterUpgrade((cluster) -> {
assertTrue(((IInvokableInstance)cluster.get(1)).callOnInstance(() -> StorageService.instance.getRangeToAddressMap("system_cluster_metadata").isEmpty()));
try
{
((IInvokableInstance) cluster.get(1)).callOnInstance(() -> StorageService.instance.getRangeToAddressMap("system_cluster_metadata").isEmpty());
fail();
}
catch (Exception e)
{
assertTrue(e.getMessage().contains("Unknown keyspace system_cluster_metadata"));
}
cluster.get(1).nodetoolResult("cms", "initialize").asserts().success();
((IInvokableInstance) cluster.get(1)).callOnInstance(() -> StorageService.instance.getRangeToAddressMap("system_cluster_metadata").isEmpty());
// make sure we can execute transformations:
cluster.schemaChange(withKeyspace("ALTER TABLE %s.tbl with comment = 'hello123'"));
}).run();

64
test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeDC2InitializeTest.java Normal file
View File

@ -0,0 +1,64 @@
/*
* 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.upgrade;
import com.google.common.collect.ImmutableMap;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.distributed.Constants;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.distributed.shared.NetworkTopology;
import static org.junit.Assert.assertFalse;
public class ClusterMetadataUpgradeDC2InitializeTest extends UpgradeTestBase
{
public static final Logger logger = LoggerFactory.getLogger(ClusterMetadataUpgradeDC2InitializeTest.class);
@Test
public void testCMSInitializeOnDC2NodeAfterUpgrade() throws Throwable
{
new TestCase()
.nodes(2)
.nodesToUpgrade(1, 2)
.withNodeIdTopology(ImmutableMap.of(1, NetworkTopology.dcAndRack("dc1", "rack1"),
2, NetworkTopology.dcAndRack("dc2", "rack2")))
.withConfig((cfg) -> cfg.with(Feature.NETWORK, Feature.GOSSIP)
.set(Constants.KEY_DTEST_FULL_STARTUP, true))
.upgradesToCurrentFrom(v41)
.setup((cluster) -> {
})
.runBeforeClusterUpgrade(cluster -> {
cluster.forEach(node -> {
node.flush("system");
});
})
.runAfterClusterUpgrade((cluster) -> {
// Run cms initialize on the node in dc2 (node 2) instead of dc1.
cluster.get(2).nodetoolResult("cms", "initialize").asserts().success();
cluster.forEach(i -> assertFalse("node " + i.config().num() + " is still in MIGRATING STATE",
ClusterUtils.isMigrating((IInvokableInstance) i)));
}).run();
}
}

2
test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeHarryTest.java
View File

@ -24,6 +24,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Ignore;
import org.junit.Test;
import org.apache.cassandra.concurrent.Interruptible;
@ -55,6 +56,7 @@ import static org.junit.Assert.assertEquals;
public class ClusterMetadataUpgradeHarryTest extends UpgradeTestBase
{
@Test
@Ignore("CASSANDRA-21473")
public void simpleUpgradeTest()
{
AtomicReference<Interruptible> executor = new AtomicReference<>();

114
test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeInconsistentPeersV2Test.java Normal file
View File

@ -0,0 +1,114 @@
/*
* 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.upgrade;
import java.util.Map;
import org.junit.Test;
import org.apache.cassandra.distributed.Constants;
import org.apache.cassandra.distributed.UpgradeableCluster;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.IUpgradeableInstance;
import org.apache.cassandra.distributed.api.SimpleQueryResult;
import org.apache.cassandra.distributed.shared.NetworkTopology;
import org.apache.cassandra.locator.MetaStrategy;
import org.apache.cassandra.locator.NetworkTopologyStrategy;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.tcm.ClusterMetadata;
import static java.lang.String.format;
import static org.apache.cassandra.schema.SchemaConstants.METADATA_KEYSPACE_NAME;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
public class ClusterMetadataUpgradeInconsistentPeersV2Test extends UpgradeTestBase
{
@Test
public void upgradeWithInconsistentSystemPeersV2Test() throws Throwable
{
new TestCase()
.nodes(3)
.withNodeIdTopology(NetworkTopology.networkTopology(3, (i) -> NetworkTopology.dcAndRack("datacenter0" + i % 2, "rack0" + i)))
.nodesToUpgrade(1, 2, 3)
.withConfig((cfg) -> cfg.with(Feature.NETWORK, Feature.GOSSIP)
.set(Constants.KEY_DTEST_FULL_STARTUP, true))
.upgradesToCurrentFrom(v41)
.setup((cluster) -> {
// insert mismatching entries into system.peers_v2 table
for (int i = 1; i <= 3; i++)
{
cluster.get(i).executeInternal(format("insert into system.peers_v2 (peer, peer_port, data_center) " +
"values ('10.10.10.10', 7000, 'a%s')", i));
cluster.get(i).flush("system");
}
})
.runAfterClusterUpgrade((cluster) -> {
// The system_cluster_metadata keyspace shouldn't be created until the CMS is intialized
assertNoMetaKeyspace(cluster);
cluster.get(3).nodetoolResult("cms", "initialize").asserts().success();
// post-initialization, the replication params for system_cluster_metadata should be RF 1 in the DC of the
// first CMS member - node3 / datacenter01
Map<String, String> actualReplication = Map.of("class", MetaStrategy.class.getName(), "datacenter01", "1");
Map<String, String> fromSystemTable = Map.of("class", NetworkTopologyStrategy.class.getName(), "datacenter01", "1");
assertReplicationParams(cluster, actualReplication, fromSystemTable);
}).run();
}
private static void assertNoMetaKeyspace(UpgradeableCluster cluster)
{
for (IUpgradeableInstance inst : cluster)
{
IInvokableInstance i = (IInvokableInstance) inst;
boolean found = i.callOnInstance(() -> ClusterMetadata.current().schema.getKeyspaces().containsKeyspace(METADATA_KEYSPACE_NAME));
assertFalse("Metadata keyspace present on node" + i + " when it should not be", found);
SimpleQueryResult res = inst.executeInternalWithResult("select replication from system_schema.keyspaces " +
"where keyspace_name = ?",
METADATA_KEYSPACE_NAME);
assertFalse(res.hasNext());
}
}
private static void assertReplicationParams(UpgradeableCluster cluster,
Map<String, String> expectedActual,
Map<String, String> expectedInSystemTable)
{
for (IUpgradeableInstance inst : cluster)
{
IInvokableInstance i = (IInvokableInstance) inst;
Map<String, String> rs = i.callOnInstance(() -> {
ReplicationParams r = ClusterMetadata.current().schema.getKeyspaceMetadata("system_cluster_metadata").params.replication;
return r.asMap();
});
assertEquals(rs, expectedActual);
SimpleQueryResult res = inst.executeInternalWithResult("select replication from system_schema.keyspaces " +
"where keyspace_name = ?",
METADATA_KEYSPACE_NAME);
assertTrue(res.hasNext());
Map<String, String> replication = res.next().get("replication");
assertEquals(replication, expectedInSystemTable);
}
}
}

109
test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradePeersHostIdsTest.java Normal file
View File

@ -0,0 +1,109 @@
/*
* 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.upgrade;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import org.junit.Test;
import org.apache.cassandra.distributed.Constants;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IUpgradeableInstance;
import org.apache.cassandra.distributed.api.Row;
import org.apache.cassandra.distributed.api.SimpleQueryResult;
import org.apache.cassandra.distributed.impl.TestEndpointCache;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.tcm.membership.NodeId;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
public class ClusterMetadataUpgradePeersHostIdsTest extends UpgradeTestBase
{
@Test
public void upgradeHostIdUpdateTest() throws Throwable
{
Map<InetAddressAndPort, UUID> preUpgradeIDs = new HashMap<>();
new TestCase()
.nodes(3)
.nodesToUpgrade(1, 2, 3)
.withConfig((cfg) -> cfg.with(Feature.NETWORK, Feature.GOSSIP)
.set(Constants.KEY_DTEST_FULL_STARTUP, true))
.upgradesToCurrentFrom(v41)
.setup((cluster) -> {
// combine the system.peers_v2 entries from each instance pre-upgrade
for (int i = 1; i <= 3; i++)
preUpgradeIDs.putAll(getHostIdsFromSystemPeersV2(cluster.get(i)));
assertEquals(3, preUpgradeIDs.size());
for (UUID hostId : preUpgradeIDs.values())
assertFalse(NodeId.isValidNodeId(hostId));
})
.runAfterClusterUpgrade((cluster) -> {
for (int i = 1; i <= 3; i++)
{
// system.peers/peers_v2 should still contain the pre-upgrade ids
IUpgradeableInstance inst = cluster.get(i);
Map<InetAddressAndPort, UUID> expected = new HashMap<>(preUpgradeIDs);
expected.remove(TestEndpointCache.toCassandraInetAddressAndPort(inst.config().broadcastAddress()));
assertEquals(expected, getHostIdsFromSystemPeersV2(cluster.get(i)));
}
// initialize the CMS and fetch new ids from the ClusterMetadata Directory
cluster.get(1).nodetoolResult("cms", "initialize").asserts().success();
Map<InetAddressAndPort, UUID> postUpgradeIDs = getHostIdsFromClusterMetadata(cluster.get(1));
for (int i = 1; i <= 3; i++)
{
// assert system.peers/peers_v2 now contain the post-upgrade ids
IUpgradeableInstance inst = cluster.get(i);
Map<InetAddressAndPort, UUID> expected = new HashMap<>(postUpgradeIDs);
expected.remove(TestEndpointCache.toCassandraInetAddressAndPort(inst.config().broadcastAddress()));
assertEquals(expected, getHostIdsFromSystemPeersV2(cluster.get(i)));
}
}).run();
}
private static Map<InetAddressAndPort, UUID> getHostIdsFromSystemPeersV2(IUpgradeableInstance instance)
{
Map<InetAddressAndPort, UUID> hostIds = new HashMap<>();
SimpleQueryResult res = instance.executeInternalWithResult("select peer, peer_port, host_id from system.peers_v2");
while(res.hasNext())
{
Row row = res.next();
hostIds.put(InetAddressAndPort.getByAddressOverrideDefaults(row.get(0), row.getInteger(1)), row.getUUID(2));
}
return hostIds;
}
private static Map<InetAddressAndPort, UUID> getHostIdsFromClusterMetadata(IUpgradeableInstance instance)
{
Map<InetAddressAndPort, UUID> hostIds = new HashMap<>();
SimpleQueryResult res = instance.executeInternalWithResult("select broadcast_address, broadcast_port, host_id from system_views.cluster_metadata_directory");
while(res.hasNext())
{
Row row = res.next();
hostIds.put(InetAddressAndPort.getByAddressOverrideDefaults(row.get(0), row.getInteger(1)), row.getUUID(2));
}
return hostIds;
}
}

10
test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeTest.java
View File

@ -56,9 +56,9 @@ public class ClusterMetadataUpgradeTest extends UpgradeTestBase
})
.runAfterClusterUpgrade((cluster) -> {
Object [][] r = cluster.get(1).executeInternal("select keyspace_name from system_schema.keyspaces where keyspace_name='system_cluster_metadata'");
assertEquals(1, r.length);
assertEquals(0, r.length);
r = cluster.get(1).executeInternal("select table_name from system_schema.tables where keyspace_name='system_cluster_metadata' and table_name='distributed_metadata_log'");
assertEquals(1, r.length);
assertEquals(0, r.length);
cluster.get(1).nodetoolResult("cms","initialize").asserts().success();
cluster.forEach(i ->
@ -67,6 +67,12 @@ public class ClusterMetadataUpgradeTest extends UpgradeTestBase
assertFalse("node " + i.config().num() + " is still in MIGRATING STATE",
ClusterUtils.isMigrating((IInvokableInstance) i));
});
r = cluster.get(1).executeInternal("select keyspace_name from system_schema.keyspaces where keyspace_name='system_cluster_metadata'");
assertEquals(1, r.length);
r = cluster.get(1).executeInternal("select table_name from system_schema.tables where keyspace_name='system_cluster_metadata' and table_name='distributed_metadata_log'");
assertEquals(1, r.length);
cluster.get(2).nodetoolResult("cms", "reconfigure", "3").asserts().success();
cluster.schemaChange(withKeyspace("create table %s.xyz (id int primary key)"));
cluster.forEach(i -> {

163
test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeUnexpectedFailureTest.java
View File

@ -18,7 +18,6 @@
package org.apache.cassandra.distributed.upgrade;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
@ -30,39 +29,124 @@ import net.bytebuddy.implementation.bind.annotation.SuperCall;
import org.junit.Test;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.distributed.Constants;
import org.apache.cassandra.distributed.UpgradeableCluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.IUpgradeableInstance;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.tcm.migration.Election;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.log.LocalLog;
import org.apache.cassandra.tcm.migration.CMSInitializationException;
import org.apache.cassandra.utils.Shared;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.cassandra.schema.SchemaConstants.METADATA_KEYSPACE_NAME;
import static org.apache.cassandra.schema.SchemaConstants.SCHEMA_KEYSPACE_NAME;
import static org.apache.cassandra.schema.SchemaKeyspaceTables.COLUMNS;
import static org.apache.cassandra.schema.SchemaKeyspaceTables.KEYSPACES;
import static org.apache.cassandra.schema.SchemaKeyspaceTables.TABLES;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
public class ClusterMetadataUpgradeUnexpectedFailureTest extends UpgradeTestBase
{
@Test
public void upgradeFailsUnexpectedlyAfterInitiation() throws Throwable
public void upgradeFailsUnexpectedlyBeforePreInitialize() throws Throwable
{
testCMSInitializationError(true, false, "Something unexpected went wrong");
}
@Test
public void upgradeFailsUnexpectedlyBeforeInitialize() throws Throwable
{
testCMSInitializationError(false, true, "CMS initialization failed, see logs for details");
}
private void testCMSInitializationError(boolean failBeforePreInit,
boolean failAfterPreInit,
String expectedError) throws Throwable
{
Consumer<UpgradeableCluster.Builder > builderUpdater = builder -> builder.withInstanceInitializer(BBInstaller::installUpgradeVersionBB);
new TestCase()
.nodes(3)
.nodesToUpgrade(1, 2, 3)
.withConfig((cfg) -> cfg.with(Feature.NETWORK, Feature.GOSSIP)
.set(Constants.KEY_DTEST_FULL_STARTUP, true))
.upgradesToCurrentFrom(v41)
.withBuilder(builderUpdater)
.setup((cluster) -> {
cluster.schemaChange(withKeyspace("ALTER KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor':2}"));
cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
})
.runAfterClusterUpgrade((cluster) -> {
// we injected a BB helper to trigger an unexpected failure on the first attempt at initialization.
// i.e. an exception that must be caught as opposed to a mismatch in metadata or down peer.
cluster.get(1).nodetoolResult("cms", "initialize").asserts().failure().errorContains("Something unexpected went wrong");
// handling the failure should have included cleaning up any state so that another attempt can be
// made, which this time should succeed.
cluster.get(1).nodetoolResult("cms", "initialize").asserts().success();
}).run();
.nodes(3)
.nodesToUpgrade(1, 2, 3)
.withConfig((cfg) -> cfg.with(Feature.NETWORK, Feature.GOSSIP)
.set(Constants.KEY_DTEST_FULL_STARTUP, true))
.upgradesToCurrentFrom(v41)
.withBuilder(builderUpdater)
.setup((cluster) -> {
BBState.failBeforePreInitialize.set(failBeforePreInit);
BBState.failAfterPreInitialize.set(failAfterPreInit);
cluster.schemaChange(withKeyspace("ALTER KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor':2}"));
cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
for (int i = 0; i < 10; i++)
{
cluster.get(1).coordinator().execute("INSERT into " + KEYSPACE + ".tbl (pk, ck, v) VALUES (?, ?, ?)",
ConsistencyLevel.ALL,
i, i, i);
}
})
.runAfterClusterUpgrade((cluster) -> {
String oldHostId = getHostId(cluster.get(1));
// we injected a BB helper to trigger an unexpected failure on the first attempt at initialization.
// i.e. an exception that must be caught as opposed to a mismatch in metadata or down peer.
cluster.get(1).nodetoolResult("cms", "initialize").asserts().failure().errorContains(expectedError);
assertEquals(oldHostId, getHostId(cluster.get(1)));
// handling the failure should have included cleaning up any state so that another attempt can be
// made, which this time should succeed.
for(IUpgradeableInstance inst : cluster)
inst.nodetoolResult("cms").asserts().success().stdoutContains("Service State: GOSSIP");
// Basic smoke test
for (int i = 0; i < 10; i++)
{
Object [][] rows = cluster.get(1)
.coordinator()
.execute("SELECT v from " + KEYSPACE + ".tbl WHERE pk = ?",
ConsistencyLevel.ALL, i);
assertEquals(1, rows.length);
assertEquals(i, rows[0][0]);
}
// Make sure that no trace of the metadata keyspace is present after the CMS initialization failure
assertSchemaTablesContent(cluster, true);
Object[][] rows = cluster.get(1).coordinator().execute("DESCRIBE FULL SCHEMA", ConsistencyLevel.NODE_LOCAL);
for (Object[] row : rows)
assertFalse(row[0].toString().equalsIgnoreCase(METADATA_KEYSPACE_NAME));
assertEquals(oldHostId, getHostId(cluster.get(1)));
// A subsequent initialization should succeed
cluster.get(1).nodetoolResult("cms", "initialize").asserts().success();
assertNotEquals(oldHostId, getHostId(cluster.get(1)));
assertSchemaTablesContent(cluster, false);
}).run();
}
private static void assertSchemaTablesContent(UpgradeableCluster cluster, boolean expectEmpty)
{
for (String schemaTable : new String[] { KEYSPACES, TABLES, COLUMNS })
{
Object[][] rows = cluster.get(1)
.coordinator()
.execute("SELECT * FROM " + SCHEMA_KEYSPACE_NAME + "." + schemaTable + " WHERE keyspace_name = ?" ,
ConsistencyLevel.ALL, METADATA_KEYSPACE_NAME);
if (expectEmpty)
assertEquals(0, rows.length);
else
assertTrue(rows.length >= 1);
}
}
private static String getHostId(IUpgradeableInstance i)
{
return ((IInvokableInstance)i).callOnInstance(() -> SystemKeyspace.getLocalHostId().toString());
}
public static class BBInstaller
@ -71,8 +155,15 @@ public class ClusterMetadataUpgradeUnexpectedFailureTest extends UpgradeTestBase
{
try
{
new ByteBuddy().rebase(Election.class)
.method(named("finish"))
// Fail before the LocalLog is initialized with the PRE_INITIALIZE_CMS
new ByteBuddy().rebase(LocalLog.class)
.method(named("bootstrap"))
.intercept(MethodDelegation.to(BBInterceptor.class))
.make()
.load(classLoader, ClassLoadingStrategy.Default.INJECTION);
// Fail after the PRE_INITIALIZE_CMS has been enacted, as the INITIALIZE_CMS is being committed
new ByteBuddy().rebase(ClusterMetadataService.class)
.method(named("commit"))
.intercept(MethodDelegation.to(BBInterceptor.class))
.make()
.load(classLoader, ClassLoadingStrategy.Default.INJECTION);
@ -88,14 +179,20 @@ public class ClusterMetadataUpgradeUnexpectedFailureTest extends UpgradeTestBase
}
}
@Shared
public static class BBState
{
public static AtomicBoolean failBeforePreInitialize = new AtomicBoolean(false);
public static AtomicBoolean failAfterPreInitialize = new AtomicBoolean(false);
}
public static class BBInterceptor
{
private static final AtomicBoolean firstAttempt = new AtomicBoolean(true);
@SuppressWarnings("unused")
public static void finish(Set<InetAddressAndPort> sendTo, @SuperCall Callable<Void> zuper)
public static void bootstrap(InetAddressAndPort addr, String datacenterSet, @SuperCall Callable<Void> zuper)
{
if (firstAttempt.getAndSet(false))
if (BBState.failBeforePreInitialize.getAndSet(false))
throw new IllegalStateException("Something unexpected went wrong");
try
@ -107,5 +204,19 @@ public class ClusterMetadataUpgradeUnexpectedFailureTest extends UpgradeTestBase
throw new RuntimeException(e);
}
}
public static ClusterMetadata commit(@SuperCall Callable<ClusterMetadata> zuper)
{
if (BBState.failAfterPreInitialize.getAndSet(false))
throw new CMSInitializationException();
try
{
return zuper.call();
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
}
}

8
test/unit/org/apache/cassandra/ServerTestUtils.java
View File

@ -274,13 +274,7 @@ public final class ServerTestUtils
public static void initCMS()
{
// Effectively disable automatic snapshots using AtomicLongBackedProcessor and LocaLLog.Sync interacts
// badly with submitting SealPeriod transformations from the log listener. In this configuration, SealPeriod
// commits performed on NonPeriodicTasks threads end up actually performing the transformations as well as
// calling the pre and post commit listeners, which is not threadsafe. In a non-test setup the processing of
// log entries is always done by the dedicated log follower thread.
DatabaseDescriptor.setMetadataSnapshotFrequency(Integer.MAX_VALUE);
IPartitioner partitioner = DatabaseDescriptor.getPartitioner();
Location location = DatabaseDescriptor.getLocator().local();
boolean addListeners = true;
@ -304,7 +298,7 @@ public final class ServerTestUtils
ClusterMetadataService.setInstance(service);
log.readyUnchecked();
log.bootstrap(FBUtilities.getBroadcastAddressAndPort(), location.datacenter);
log.bootstrap(FBUtilities.getBroadcastAddressAndPort(), location.datacenter, (p) -> {});
service.commit(new Initialize(ClusterMetadata.current()));
QueryProcessor.registerStatementInvalidatingListener();
service.mark();

5
test/unit/org/apache/cassandra/schema/DistributedSchemaTest.java
View File

@ -19,14 +19,12 @@
package org.apache.cassandra.schema;
import java.util.HashMap;
import java.util.Set;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.auth.AuthKeyspace;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.service.reads.PercentileSpeculativeRetryPolicy;
import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
@ -64,8 +62,7 @@ public class DistributedSchemaTest
KeyspaceParams.simple(3),
Tables.of(modifiedTable, nonstandard));
DistributedSchema schema = DistributedSchema.fromSystemTables(Keyspaces.of(km),
Set.of(DatabaseDescriptor.getLocalDataCenter()));
DistributedSchema schema = DistributedSchema.fromSystemTables(Keyspaces.of(km));
KeyspaceMetadata merged = schema.getKeyspaceMetadata(SchemaConstants.AUTH_KEYSPACE_NAME);
assertThat(merged.getTableOrViewNullable(AuthKeyspace.ROLES)).isEqualTo(modifiedTable);
assertThat(merged.getTableOrViewNullable("nonstandard")).isEqualTo(nonstandard);

7
test/unit/org/apache/cassandra/tcm/log/DistributedLogStateTest.java
View File

@ -35,6 +35,7 @@ import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.MetadataSnapshots;
import org.apache.cassandra.tcm.transformations.CustomTransformation;
import org.apache.cassandra.tcm.transformations.TriggerSnapshot;
import org.apache.cassandra.tcm.transformations.cms.PreInitialize;
import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
import static org.apache.cassandra.db.ColumnFamilyStore.FlushReason.UNIT_TESTS;
@ -76,8 +77,12 @@ public class DistributedLogStateTest extends LogStateTestBase
}
@Override
public void insertRegularEntry()
public void insertRegularEntry() throws IOException
{
if (currentEpoch == Epoch.FIRST)
DistributedMetadataLogKeyspace.insertPreInitialize(PreInitialize.blank());
nextEpoch = currentEpoch.nextEpoch();
boolean applied = DistributedMetadataLogKeyspace.tryCommit(new Entry.Id(currentEpoch.getEpoch()),
CustomTransformation.make((int) currentEpoch.getEpoch()),

2
test/unit/org/apache/cassandra/tcm/log/LogListenerNotificationTest.java
View File

@ -101,7 +101,7 @@ public class LogListenerNotificationTest
.withLogListener(listener)
.createLog();
log.readyUnchecked();
log.append(new Entry(Entry.Id.NONE, Epoch.FIRST, PreInitialize.forTesting()));
log.append(new Entry(Entry.Id.NONE, Epoch.FIRST, PreInitialize.blank()));
log.append(input);
}

2
test/unit/org/apache/cassandra/utils/CassandraGenerators.java
View File

@ -1946,7 +1946,7 @@ public final class CassandraGenerators
Epoch epoch = epochGen.generate(rnd);
IPartitioner partitioner = partitionerGen.generate(rnd);
Directory directory = Directory.EMPTY;
DistributedSchema schema = DistributedSchema.first(directory.knownDatacenters());
DistributedSchema schema = DistributedSchema.empty();
TokenMap tokenMap = new TokenMap(partitioner);
DataPlacements placements = DataPlacements.EMPTY;
AccordFastPath accordFastPath = accordFastPathGen.generate(rnd);