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

Command to Exclude Replicas from Durability Status Coordination 

patch by Caleb Rackliffe; reviewed by David Capwell, Sam Tunnicliffe, and Benedict Elliott Smith for CASSANDRA-19321
This commit is contained in:
Caleb Rackliffe 2024-07-16 17:40:49 -05:00 committed by David Capwell
parent 325e48ac39
commit 2b01b5fa79
28 changed files with 1047 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit 449b2b4d0bf4bb44d55a3c57f712a4d5a15e7220
Subproject commit 81c02769f9ad73ef3aba0675c2217fc74b8a4a4c

2
src/java/org/apache/cassandra/service/CassandraDaemon.java
View File

@ -78,6 +78,7 @@ import org.apache.cassandra.net.StartupClusterConnectivityChecker;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.security.ThreadAwareSecurityManager;
import org.apache.cassandra.service.accord.AccordOperations;
import org.apache.cassandra.service.paxos.PaxosState;
import org.apache.cassandra.service.snapshot.SnapshotManager;
import org.apache.cassandra.streaming.StreamManager;
@ -269,6 +270,7 @@ public class CassandraDaemon
Startup.initialize(DatabaseDescriptor.getSeeds());
disableAutoCompaction(Schema.instance.distributedKeyspaces().names());
CMSOperations.initJmx();
AccordOperations.initJmx();
if (ClusterMetadata.current().myNodeId() != null)
RegistrationStatus.instance.onRegistration();
}

76
src/java/org/apache/cassandra/service/accord/AccordOperations.java Normal file
View File

@ -0,0 +1,76 @@
/*
* 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.service.accord;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.transformations.AccordMarkStale;
import org.apache.cassandra.tcm.transformations.AccordMarkRejoining;
import org.apache.cassandra.utils.MBeanWrapper;
public class AccordOperations implements AccordOperationsMBean
{
public static final String MBEAN_OBJECT_NAME = "org.apache.cassandra.service.accord:type=AccordOperations";
public static final AccordOperations instance = new AccordOperations(ClusterMetadataService.instance());
private final ClusterMetadataService cms;
public static void initJmx()
{
MBeanWrapper.instance.registerMBean(instance, MBEAN_OBJECT_NAME);
}
private AccordOperations(ClusterMetadataService cms)
{
this.cms = cms;
}
@Override
public Map<String, String> describe()
{
Map<String, String> info = new HashMap<>();
ClusterMetadata metadata = ClusterMetadata.current();
info.put("EPOCH", Long.toString(metadata.epoch.getEpoch()));
String staleReplicas = metadata.accordStaleReplicas.ids().stream().sorted().map(Object::toString).collect(Collectors.joining(","));
info.put("STALE_REPLICAS", staleReplicas);
return info;
}
@Override
public void accordMarkStale(List<String> nodeIdStrings)
{
Set<NodeId> nodeIds = nodeIdStrings.stream().map(NodeId::fromString).collect(Collectors.toSet());
cms.commit(new AccordMarkStale(nodeIds));
}
@Override
public void accordMarkRejoining(List<String> nodeIdStrings)
{
Set<NodeId> nodeIds = nodeIdStrings.stream().map(NodeId::fromString).collect(Collectors.toSet());
cms.commit(new AccordMarkRejoining(nodeIds));
}
}

31
src/java/org/apache/cassandra/service/accord/AccordOperationsMBean.java Normal file
View File

@ -0,0 +1,31 @@
/*
* 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.service.accord;
import java.util.List;
import java.util.Map;
public interface AccordOperationsMBean
{
Map<String, String> describe();
void accordMarkStale(List<String> nodeIds);
void accordMarkRejoining(List<String> nodeIds);
}

137
src/java/org/apache/cassandra/service/accord/AccordStaleReplicas.java Normal file
View File

@ -0,0 +1,137 @@
/*
* 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.service.accord;
import java.io.IOException;
import java.util.Objects;
import java.util.Set;
import javax.annotation.concurrent.Immutable;
import com.google.common.collect.ImmutableSet;
import accord.local.Node;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.service.accord.serializers.TopologySerializers;
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.CollectionSerializers;
@Immutable
public class AccordStaleReplicas implements MetadataValue<AccordStaleReplicas>
{
public static final AccordStaleReplicas EMPTY = new AccordStaleReplicas(ImmutableSet.of(), Epoch.EMPTY);
private final Set<Node.Id> staleIds;
private final Epoch lastModified;
AccordStaleReplicas(Set<Node.Id> staleIds, Epoch lastModified)
{
this.staleIds = staleIds;
this.lastModified = lastModified;
}
@Override
public AccordStaleReplicas withLastModified(Epoch epoch)
{
return new AccordStaleReplicas(staleIds, epoch);
}
@Override
public Epoch lastModified()
{
return lastModified;
}
public AccordStaleReplicas withNodeIds(Set<Node.Id> ids)
{
ImmutableSet.Builder<Node.Id> builder = new ImmutableSet.Builder<>();
Set<Node.Id> newIds = builder.addAll(staleIds).addAll(ids).build();
return new AccordStaleReplicas(newIds, lastModified);
}
public AccordStaleReplicas without(Set<Node.Id> ids)
{
ImmutableSet.Builder<Node.Id> builder = new ImmutableSet.Builder<>();
for (Node.Id staleId : staleIds)
if (!ids.contains(staleId))
builder.add(staleId);
return new AccordStaleReplicas(builder.build(), lastModified);
}
public boolean contains(Node.Id nodeId)
{
return staleIds.contains(nodeId);
}
public Set<Node.Id> ids()
{
return staleIds;
}
@Override
public String toString()
{
return "AccordStaleReplicas{staleIds=" + staleIds + ", lastModified=" + lastModified + '}';
}
@Override
public boolean equals(Object o)
{
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
AccordStaleReplicas that = (AccordStaleReplicas) o;
return Objects.equals(staleIds, that.staleIds) && Objects.equals(lastModified, that.lastModified);
}
@Override
public int hashCode()
{
return Objects.hash(staleIds, lastModified);
}
public static final MetadataSerializer<AccordStaleReplicas> serializer = new MetadataSerializer<>()
{
@Override
public void serialize(AccordStaleReplicas replicas, DataOutputPlus out, Version version) throws IOException
{
CollectionSerializers.serializeCollection(replicas.staleIds, out, version, TopologySerializers.nodeId);
Epoch.serializer.serialize(replicas.lastModified, out, version);
}
@Override
public AccordStaleReplicas deserialize(DataInputPlus in, Version version) throws IOException
{
return new AccordStaleReplicas(CollectionSerializers.deserializeSet(in, version, TopologySerializers.nodeId),
Epoch.serializer.deserialize(in, version));
}
@Override
public long serializedSize(AccordStaleReplicas replicas, Version version)
{
return CollectionSerializers.serializedCollectionSize(replicas.staleIds, version, TopologySerializers.nodeId)
+ Epoch.serializer.serializedSize(replicas.lastModified, version);
}
};
}

25
src/java/org/apache/cassandra/service/accord/AccordTopology.java
View File

@ -78,7 +78,7 @@ public class AccordTopology
}
}
static class KeyspaceShard
public static class KeyspaceShard
{
private final KeyspaceMetadata keyspace;
private final Range<Token> range;
@ -139,7 +139,7 @@ public class AccordTopology
return new KeyspaceShard(keyspace, range, nodes, pending);
}
public static List<KeyspaceShard> forKeyspace(KeyspaceMetadata keyspace, DataPlacements placements, Directory directory, ShardLookup lookup)
public static List<KeyspaceShard> forKeyspace(KeyspaceMetadata keyspace, DataPlacements placements, Directory directory)
{
ReplicationParams replication = keyspace.params.replication;
DataPlacement placement = placements.get(replication);
@ -154,6 +154,16 @@ public class AccordTopology
}
return shards;
}
public List<Node.Id> nodes()
{
return nodes;
}
public Range<Token> range()
{
return range;
}
}
static TokenRange minRange(TableId table, Token token)
@ -219,7 +229,9 @@ public class AccordTopology
return builder.build();
}
public static Topology createAccordTopology(Epoch epoch, DistributedSchema schema, DataPlacements placements, Directory directory, AccordFastPath accordFastPath, ShardLookup lookup)
public static Topology createAccordTopology(Epoch epoch, DistributedSchema schema, DataPlacements placements,
Directory directory, AccordFastPath accordFastPath, ShardLookup lookup,
AccordStaleReplicas staleReplicas)
{
List<Shard> shards = new ArrayList<>();
Set<Node.Id> unavailable = accordFastPath.unavailableIds();
@ -230,17 +242,18 @@ public class AccordTopology
List<TableMetadata> tables = keyspace.tables.stream().filter(TableMetadata::requiresAccordSupport).collect(Collectors.toList());
if (tables.isEmpty())
continue;
List<KeyspaceShard> ksShards = KeyspaceShard.forKeyspace(keyspace, placements, directory, lookup);
List<KeyspaceShard> ksShards = KeyspaceShard.forKeyspace(keyspace, placements, directory);
tables.forEach(table -> ksShards.forEach(shard -> shards.add(shard.createForTable(table, unavailable, dcMap, lookup))));
}
shards.sort((a, b) -> a.range.compare(b.range));
return new Topology(epoch.getEpoch(), shards.toArray(new Shard[0]));
return new Topology(epoch.getEpoch(), staleReplicas.ids(), shards.toArray(new Shard[0]));
}
public static Topology createAccordTopology(ClusterMetadata metadata, ShardLookup lookup)
{
return createAccordTopology(metadata.epoch, metadata.schema, metadata.placements, metadata.directory, metadata.accordFastPath, lookup);
return createAccordTopology(metadata.epoch, metadata.schema, metadata.placements, metadata.directory, metadata.accordFastPath, lookup, metadata.accordStaleReplicas);
}
public static Topology createAccordTopology(ClusterMetadata metadata, Topology current)

5
src/java/org/apache/cassandra/service/accord/serializers/TopologySerializers.java
View File

@ -159,6 +159,7 @@ public class TopologySerializers
{
out.writeLong(topology.epoch());
ArraySerializers.serializeArray(topology.unsafeGetShards(), out, version, shard);
CollectionSerializers.serializeCollection(topology.staleIds(), out, version, TopologySerializers.nodeId);
}
@Override
@ -166,7 +167,8 @@ public class TopologySerializers
{
long epoch = in.readLong();
Shard[] shards = ArraySerializers.deserializeArray(in, version, shard, Shard[]::new);
return new Topology(epoch, shards);
Set<Node.Id> staleIds = CollectionSerializers.deserializeSet(in, version, TopologySerializers.nodeId);
return new Topology(epoch, staleIds, shards);
}
@Override
@ -175,6 +177,7 @@ public class TopologySerializers
long size = 0;
size += TypeSizes.LONG_SIZE; // epoch
size += ArraySerializers.serializedArraySize(topology.unsafeGetShards(), version, shard);
size += CollectionSerializers.serializedCollectionSize(topology.staleIds(), version, TopologySerializers.nodeId);
return size;
}
};

80
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;
@ -57,6 +58,8 @@ import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.Keyspaces;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.service.accord.AccordStaleReplicas;
import org.apache.cassandra.service.accord.AccordTopology;
import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState;
import org.apache.cassandra.service.consensus.migration.TableMigrationState;
import org.apache.cassandra.tcm.extensions.ExtensionKey;
@ -100,6 +103,7 @@ public class ClusterMetadata
public final InProgressSequences inProgressSequences;
public final ConsensusMigrationState consensusMigrationState;
public final ImmutableMap<ExtensionKey<?,?>, ExtensionValue<?>> extensions;
public final AccordStaleReplicas accordStaleReplicas;
// This isn't serialized as part of ClusterMetadata it's really just a view over the Directory.
public final Locator locator;
@ -135,7 +139,8 @@ public class ClusterMetadata
LockedRanges.EMPTY,
InProgressSequences.EMPTY,
ConsensusMigrationState.EMPTY,
ImmutableMap.of());
ImmutableMap.of(),
AccordStaleReplicas.EMPTY);
}
public ClusterMetadata(Epoch epoch,
@ -148,7 +153,8 @@ public class ClusterMetadata
LockedRanges lockedRanges,
InProgressSequences inProgressSequences,
ConsensusMigrationState consensusMigrationState,
Map<ExtensionKey<?, ?>, ExtensionValue<?>> extensions)
Map<ExtensionKey<?, ?>, ExtensionValue<?>> extensions,
AccordStaleReplicas accordStaleReplicas)
{
this(EMPTY_METADATA_IDENTIFIER,
epoch,
@ -161,7 +167,8 @@ public class ClusterMetadata
lockedRanges,
inProgressSequences,
consensusMigrationState,
extensions);
extensions,
accordStaleReplicas);
}
private ClusterMetadata(int metadataIdentifier,
@ -175,7 +182,8 @@ public class ClusterMetadata
LockedRanges lockedRanges,
InProgressSequences inProgressSequences,
ConsensusMigrationState consensusMigrationState,
Map<ExtensionKey<?, ?>, ExtensionValue<?>> extensions)
Map<ExtensionKey<?, ?>, ExtensionValue<?>> extensions,
AccordStaleReplicas accordStaleReplicas)
{
// TODO: token map is a feature of the specific placement strategy, and so may not be a relevant component of
// ClusterMetadata in the long term. We need to consider how the actual components of metadata can be evolved
@ -194,16 +202,17 @@ public class ClusterMetadata
this.consensusMigrationState = consensusMigrationState;
this.extensions = ImmutableMap.copyOf(extensions);
this.locator = Locator.usingDirectory(directory);
this.accordStaleReplicas = accordStaleReplicas;
}
public ClusterMetadata withDirectory(Directory directory)
{
return new ClusterMetadata(epoch, partitioner, schema, directory, tokenMap, placements, accordFastPath, lockedRanges, inProgressSequences, consensusMigrationState, extensions);
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);
return new ClusterMetadata(epoch, partitioner, schema, directory, tokenMap, placements, accordFastPath, lockedRanges, inProgressSequences, consensusMigrationState, extensions, accordStaleReplicas);
}
public Set<InetAddressAndPort> fullCMSMembers()
@ -259,7 +268,8 @@ public class ClusterMetadata
capLastModified(lockedRanges, epoch),
capLastModified(inProgressSequences, epoch),
capLastModified(consensusMigrationState, epoch),
capLastModified(extensions, epoch));
capLastModified(extensions, epoch),
capLastModified(accordStaleReplicas, epoch));
}
public ClusterMetadata initializeClusterIdentifier(int clusterIdentifier)
@ -281,7 +291,8 @@ public class ClusterMetadata
lockedRanges,
inProgressSequences,
consensusMigrationState,
extensions);
extensions,
accordStaleReplicas);
}
private static Map<ExtensionKey<?,?>, ExtensionValue<?>> capLastModified(Map<ExtensionKey<?,?>, ExtensionValue<?>> original, Epoch maxEpoch)
@ -410,6 +421,7 @@ public class ClusterMetadata
private ConsensusMigrationState consensusMigrationState;
private final Map<ExtensionKey<?, ?>, ExtensionValue<?>> extensions;
private final Set<MetadataKey> modifiedKeys;
private AccordStaleReplicas accordStaleReplicas;
private Transformer(ClusterMetadata metadata, Epoch epoch)
{
@ -426,6 +438,7 @@ public class ClusterMetadata
this.consensusMigrationState = metadata.consensusMigrationState;
extensions = new HashMap<>(metadata.extensions);
modifiedKeys = new HashSet<>();
accordStaleReplicas = metadata.accordStaleReplicas;
}
public Epoch epoch()
@ -456,6 +469,11 @@ public class ClusterMetadata
directory = directory.withoutRackAndDC(nodeId).without(nodeId);
if (!tokenMap.tokens(nodeId).isEmpty())
tokenMap = tokenMap.unassignTokens(nodeId);
Node.Id accordId = AccordTopology.tcmIdToAccord(nodeId);
if (accordStaleReplicas.contains(accordId))
accordStaleReplicas = accordStaleReplicas.without(Collections.singleton(accordId));
return this;
}
@ -523,6 +541,11 @@ public class ClusterMetadata
directory = directory.without(replaced)
.withRackAndDC(replacement)
.withNodeState(replacement, NodeState.JOINED);
Node.Id accordId = AccordTopology.tcmIdToAccord(replaced);
if (accordStaleReplicas.contains(accordId))
accordStaleReplicas = accordStaleReplicas.without(Collections.singleton(accordId));
return this;
}
@ -551,6 +574,18 @@ public class ClusterMetadata
accordFastPath = accordFastPath.withNodeStatusSince(node, status, updateTimeMillis, updateDelayMillis);
return this;
}
public Transformer markStaleReplicas(Set<Node.Id> ids)
{
accordStaleReplicas = accordStaleReplicas.withNodeIds(ids);
return this;
}
public Transformer unmarkStaleReplicas(Set<Node.Id> ids)
{
accordStaleReplicas = accordStaleReplicas.without(ids);
return this;
}
public Transformer with(LockedRanges lockedRanges)
{
@ -677,6 +712,12 @@ public class ClusterMetadata
modifiedKeys.add(MetadataKeys.ACCORD_FAST_PATH);
accordFastPath = accordFastPath.withLastModified(epoch);
}
if (accordStaleReplicas != base.accordStaleReplicas)
{
modifiedKeys.add(MetadataKeys.ACCORD_STALE_REPLICAS);
accordStaleReplicas = accordStaleReplicas.withLastModified(epoch);
}
if (lockedRanges != base.lockedRanges)
{
@ -712,7 +753,8 @@ public class ClusterMetadata
lockedRanges,
inProgressSequences,
consensusMigrationState,
extensions),
extensions,
accordStaleReplicas),
ImmutableSet.copyOf(modifiedKeys));
}
@ -729,7 +771,8 @@ public class ClusterMetadata
lockedRanges,
inProgressSequences,
consensusMigrationState,
extensions);
extensions,
accordStaleReplicas);
}
@Override
@ -861,6 +904,7 @@ public class ClusterMetadata
lockedRanges.equals(that.lockedRanges) &&
inProgressSequences.equals(that.inProgressSequences) &&
consensusMigrationState.equals(that.consensusMigrationState) &&
accordStaleReplicas.equals(that.accordStaleReplicas) &&
extensions.equals(that.extensions);
}
@ -909,7 +953,7 @@ public class ClusterMetadata
@Override
public int hashCode()
{
return Objects.hash(epoch, schema, directory, tokenMap, placements, accordFastPath, lockedRanges, inProgressSequences, consensusMigrationState, extensions);
return Objects.hash(epoch, schema, directory, tokenMap, placements, accordFastPath, lockedRanges, inProgressSequences, consensusMigrationState, accordStaleReplicas, extensions);
}
public static ClusterMetadata current()
@ -990,7 +1034,9 @@ public class ClusterMetadata
{
AccordFastPath.serializer.serialize(metadata.accordFastPath, out, version);
ConsensusMigrationState.serializer.serialize(metadata.consensusMigrationState, out, version);
AccordStaleReplicas.serializer.serialize(metadata.accordStaleReplicas, out, version);
}
LockedRanges.serializer.serialize(metadata.lockedRanges, out, version);
InProgressSequences.serializer.serialize(metadata.inProgressSequences, out, version);
out.writeInt(metadata.extensions.size());
@ -1027,18 +1073,24 @@ public class ClusterMetadata
Directory dir = Directory.serializer.deserialize(in, version);
TokenMap tokenMap = TokenMap.serializer.deserialize(in, version);
DataPlacements placements = DataPlacements.serializer.deserialize(in, version);
AccordFastPath accordFastPath;
ConsensusMigrationState consensusMigrationState;
AccordStaleReplicas staleReplicas;
if (version.isAtLeast(V2))
{
accordFastPath = AccordFastPath.serializer.deserialize(in, version);
consensusMigrationState = ConsensusMigrationState.serializer.deserialize(in, version);
staleReplicas = AccordStaleReplicas.serializer.deserialize(in, version);
}
else
{
accordFastPath = AccordFastPath.EMPTY;
consensusMigrationState = ConsensusMigrationState.EMPTY;
staleReplicas = AccordStaleReplicas.EMPTY;
}
LockedRanges lockedRanges = LockedRanges.serializer.deserialize(in, version);
InProgressSequences ips = InProgressSequences.serializer.deserialize(in, version);
int items = in.readInt();
@ -1061,7 +1113,8 @@ public class ClusterMetadata
lockedRanges,
ips,
consensusMigrationState,
extensions);
extensions,
staleReplicas);
}
@Override
@ -1085,7 +1138,8 @@ public class ClusterMetadata
if (version.isAtLeast(V2))
{
size += AccordFastPath.serializer.serializedSize(metadata.accordFastPath, version) +
ConsensusMigrationState.serializer.serializedSize(metadata.consensusMigrationState, version);
ConsensusMigrationState.serializer.serializedSize(metadata.consensusMigrationState, version) +
AccordStaleReplicas.serializer.serializedSize(metadata.accordStaleReplicas, version);
}
size += LockedRanges.serializer.serializedSize(metadata.lockedRanges, version) +

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

@ -40,6 +40,7 @@ public class MetadataKeys
public static final MetadataKey TOKEN_MAP = make(CORE_NS, "ownership", "token_map");
public static final MetadataKey DATA_PLACEMENTS = make(CORE_NS, "ownership", "data_placements");
public static final MetadataKey ACCORD_FAST_PATH = make(CORE_NS, "ownership", "accord_fast_path");
public static final MetadataKey ACCORD_STALE_REPLICAS = make(CORE_NS, "ownership", "accord_stale_replicas");
public static final MetadataKey LOCKED_RANGES = make(CORE_NS, "sequences", "locked_ranges");
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");
@ -49,6 +50,7 @@ public class MetadataKeys
TOKEN_MAP,
DATA_PLACEMENTS,
ACCORD_FAST_PATH,
ACCORD_STALE_REPLICAS,
LOCKED_RANGES,
IN_PROGRESS_SEQUENCES,
CONSENSUS_MIGRATION_STATE);

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

@ -29,6 +29,7 @@ import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.Keyspaces;
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.Commit.Replicator;
import org.apache.cassandra.tcm.log.Entry;
@ -184,7 +185,8 @@ public class StubClusterMetadataService extends ClusterMetadataService
LockedRanges.EMPTY,
InProgressSequences.EMPTY,
ConsensusMigrationState.EMPTY,
ImmutableMap.of());
ImmutableMap.of(),
AccordStaleReplicas.EMPTY);
}
return new StubClusterMetadataService(new UniformRangePlacement(),
snapshots != null ? snapshots : MetadataSnapshots.NO_OP,

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

@ -38,6 +38,8 @@ import org.apache.cassandra.tcm.sequences.LockedRanges;
import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
import org.apache.cassandra.tcm.serialization.VerboseMetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.tcm.transformations.AccordMarkRejoining;
import org.apache.cassandra.tcm.transformations.AccordMarkStale;
import org.apache.cassandra.tcm.transformations.AlterSchema;
import org.apache.cassandra.tcm.transformations.AlterTopology;
import org.apache.cassandra.tcm.transformations.Assassinate;
@ -242,6 +244,8 @@ public interface Transformation
BEGIN_CONSENSUS_MIGRATION_FOR_TABLE_AND_RANGE(37, () -> BeginConsensusMigrationForTableAndRange.serializer),
MAYBE_FINISH_CONSENSUS_MIGRATION_FOR_TABLE_AND_RANGE(38, () -> MaybeFinishConsensusMigrationForTableAndRange.serializer),
ACCORD_MARK_STALE(39, () -> AccordMarkStale.serializer),
ACCORD_MARK_REJOINING(40, () -> AccordMarkRejoining.serializer),
;
private final Supplier<AsymmetricMetadataSerializer<Transformation, ? extends Transformation>> serializer;

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

@ -54,6 +54,7 @@ import org.apache.cassandra.schema.Keyspaces;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.SchemaKeyspace;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.accord.AccordStaleReplicas;
import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
@ -300,7 +301,8 @@ public class GossipHelper
LockedRanges.EMPTY,
InProgressSequences.EMPTY,
ConsensusMigrationState.EMPTY,
Collections.emptyMap());
Collections.emptyMap(),
AccordStaleReplicas.EMPTY);
}
public static ClusterMetadata fromEndpointStates(DistributedSchema schema, Map<InetAddressAndPort, EndpointState> epStates)
@ -389,7 +391,8 @@ public class GossipHelper
LockedRanges.EMPTY,
InProgressSequences.EMPTY,
ConsensusMigrationState.EMPTY,
extensions);
extensions,
AccordStaleReplicas.EMPTY);
DataPlacements placements = new UniformRangePlacement().calculatePlacements(Epoch.UPGRADE_GOSSIP,
forPlacementCalculation,
schema.getKeyspaces());
@ -403,7 +406,8 @@ public class GossipHelper
LockedRanges.EMPTY,
InProgressSequences.EMPTY,
ConsensusMigrationState.EMPTY,
extensions);
extensions,
AccordStaleReplicas.EMPTY);
}
public static boolean isValidForClusterMetadata(Map<InetAddressAndPort, EndpointState> epstates)

1
src/java/org/apache/cassandra/tcm/serialization/Version.java
View File

@ -37,6 +37,7 @@ public enum Version
* - Added version to PlacementForRange serializer
* - Serialize MemtableParams when serializing TableParams
* - Added AccordFastPath
* - Added AccordStaleReplicas
*/
V2(2),
/**

123
src/java/org/apache/cassandra/tcm/transformations/AccordMarkRejoining.java Normal file
View File

@ -0,0 +1,123 @@
/*
* 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 java.util.Set;
import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.local.Node;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.service.accord.AccordTopology;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Transformation;
import org.apache.cassandra.tcm.membership.NodeId;
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.utils.CollectionSerializers;
import static org.apache.cassandra.exceptions.ExceptionCode.INVALID;
public class AccordMarkRejoining implements Transformation
{
private static final Logger logger = LoggerFactory.getLogger(AccordMarkRejoining.class);
private final Set<NodeId> ids;
public AccordMarkRejoining(Set<NodeId> ids)
{
this.ids = ids;
}
@Override
public Kind kind()
{
return Kind.ACCORD_MARK_REJOINING;
}
@Override
public Result execute(ClusterMetadata prev)
{
for (NodeId id : ids)
if (!prev.directory.peerIds().contains(id))
return new Rejected(INVALID, String.format("Can not unmark node %s as it is not present in the directory.", id));
Set<Node.Id> accordIds = ids.stream().map(AccordTopology::tcmIdToAccord).collect(Collectors.toSet());
for (Node.Id id : accordIds)
if (!prev.accordStaleReplicas.contains(id))
return new Rejected(INVALID, String.format("Can not unmark node %s as it is not stale.", id));
logger.info("Unmarking " + ids + ". They will now participate in durability status coordination...");
ClusterMetadata.Transformer next = prev.transformer().unmarkStaleReplicas(accordIds);
return Transformation.success(next, LockedRanges.AffectedRanges.EMPTY);
}
@Override
public String toString()
{
return "AccordMarkRejoining{ids=" + ids + '}';
}
@Override
public boolean equals(Object o)
{
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
AccordMarkRejoining that = (AccordMarkRejoining) o;
return Objects.equals(ids, that.ids);
}
@Override
public int hashCode()
{
return Objects.hash(ids);
}
public static final AsymmetricMetadataSerializer<Transformation, AccordMarkRejoining> serializer = new AsymmetricMetadataSerializer<>()
{
@Override
public void serialize(Transformation t, DataOutputPlus out, Version version) throws IOException
{
assert t instanceof AccordMarkRejoining;
AccordMarkRejoining mark = (AccordMarkRejoining) t;
CollectionSerializers.serializeCollection(mark.ids, out, version, NodeId.serializer);
}
@Override
public AccordMarkRejoining deserialize(DataInputPlus in, Version version) throws IOException
{
return new AccordMarkRejoining(CollectionSerializers.deserializeSet(in, version, NodeId.serializer));
}
@Override
public long serializedSize(Transformation t, Version version)
{
assert t instanceof AccordMarkRejoining;
AccordMarkRejoining mark = (AccordMarkRejoining) t;
return CollectionSerializers.serializedCollectionSize(mark.ids, version, NodeId.serializer);
}
};
}

151
src/java/org/apache/cassandra/tcm/transformations/AccordMarkStale.java Normal file
View File

@ -0,0 +1,151 @@
/*
* 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.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.local.Node;
import accord.topology.Shard;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.service.accord.AccordTopology;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Transformation;
import org.apache.cassandra.tcm.membership.NodeId;
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.utils.CollectionSerializers;
import static org.apache.cassandra.exceptions.ExceptionCode.INVALID;
public class AccordMarkStale implements Transformation
{
private static final Logger logger = LoggerFactory.getLogger(AccordMarkStale.class);
private final Set<NodeId> ids;
public AccordMarkStale(Set<NodeId> ids)
{
this.ids = ids;
}
@Override
public Kind kind()
{
return Kind.ACCORD_MARK_STALE;
}
@Override
public Result execute(ClusterMetadata prev)
{
for (NodeId id : ids)
if (!prev.directory.peerIds().contains(id))
return new Rejected(INVALID, String.format("Can not mark node %s stale as it is not present in the directory.", id));
Set<Node.Id> accordIds = ids.stream().map(AccordTopology::tcmIdToAccord).collect(Collectors.toSet());
for (Node.Id id : accordIds)
if (prev.accordStaleReplicas.contains(id))
return new Rejected(INVALID, String.format("Can not mark node %s stale as it already is.", id));
for (KeyspaceMetadata keyspace : prev.schema.getKeyspaces().without(SchemaConstants.REPLICATED_SYSTEM_KEYSPACE_NAMES))
{
List<AccordTopology.KeyspaceShard> shards = AccordTopology.KeyspaceShard.forKeyspace(keyspace, prev.placements, prev.directory);
for (AccordTopology.KeyspaceShard shard : shards)
{
// We're trying to mark a node in this shard stale...
if (!Collections.disjoint(shard.nodes(), accordIds))
{
int quorumSize = Shard.slowPathQuorumSize(shard.nodes().size());
Set<Node.Id> nonStaleNodes = new HashSet<>(shard.nodes());
nonStaleNodes.removeAll(accordIds);
nonStaleNodes.removeAll(prev.accordStaleReplicas.ids());
// ...but reject the transformation if this would bring us below quorum.
if (nonStaleNodes.size() < quorumSize)
return new Rejected(INVALID, String.format("Can not mark nodes %s stale as that would leave fewer than a quorum of nodes active for range %s in keyspace '%s'.",
accordIds, shard.range(), keyspace.name));
}
}
}
logger.info("Marking " + ids + " stale. They will no longer participate in durability status coordination...");
ClusterMetadata.Transformer next = prev.transformer().markStaleReplicas(accordIds);
return Transformation.success(next, LockedRanges.AffectedRanges.EMPTY);
}
@Override
public String toString()
{
return "AccordMarkStale{ids=" + ids + '}';
}
@Override
public boolean equals(Object o)
{
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
AccordMarkStale that = (AccordMarkStale) o;
return Objects.equals(ids, that.ids);
}
@Override
public int hashCode()
{
return Objects.hash(ids);
}
public static final AsymmetricMetadataSerializer<Transformation, AccordMarkStale> serializer = new AsymmetricMetadataSerializer<>()
{
@Override
public void serialize(Transformation t, DataOutputPlus out, Version version) throws IOException
{
assert t instanceof AccordMarkStale;
AccordMarkStale mark = (AccordMarkStale) t;
CollectionSerializers.serializeCollection(mark.ids, out, version, NodeId.serializer);
}
@Override
public AccordMarkStale deserialize(DataInputPlus in, Version version) throws IOException
{
return new AccordMarkStale(CollectionSerializers.deserializeSet(in, version, NodeId.serializer));
}
@Override
public long serializedSize(Transformation t, Version version)
{
assert t instanceof AccordMarkStale;
AccordMarkStale mark = (AccordMarkStale) t;
return CollectionSerializers.serializedCollectionSize(mark.ids, version, NodeId.serializer);
}
};
}

10
src/java/org/apache/cassandra/tools/NodeProbe.java
View File

@ -120,6 +120,8 @@ import org.apache.cassandra.service.GCInspectorMXBean;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.StorageProxyMBean;
import org.apache.cassandra.service.StorageServiceMBean;
import org.apache.cassandra.service.accord.AccordOperations;
import org.apache.cassandra.service.accord.AccordOperationsMBean;
import org.apache.cassandra.streaming.StreamManagerMBean;
import org.apache.cassandra.streaming.StreamState;
import org.apache.cassandra.streaming.management.StreamStateCompositeData;
@ -154,6 +156,7 @@ public class NodeProbe implements AutoCloseable
protected StorageServiceMBean ssProxy;
protected SnapshotManagerMBean snapshotProxy;
protected CMSOperationsMBean cmsProxy;
protected AccordOperationsMBean accordProxy;
protected GossiperMBean gossProxy;
protected MemoryMXBean memProxy;
protected GCInspectorMXBean gcProxy;
@ -270,6 +273,8 @@ public class NodeProbe implements AutoCloseable
snapshotProxy = JMX.newMBeanProxy(mbeanServerConn, name, SnapshotManagerMBean.class);
name = new ObjectName(CMSOperations.MBEAN_OBJECT_NAME);
cmsProxy = JMX.newMBeanProxy(mbeanServerConn, name, CMSOperationsMBean.class);
name = new ObjectName(AccordOperations.MBEAN_OBJECT_NAME);
accordProxy = JMX.newMBeanProxy(mbeanServerConn, name, AccordOperationsMBean.class);
name = new ObjectName(MessagingService.MBEAN_NAME);
msProxy = JMX.newMBeanProxy(mbeanServerConn, name, MessagingServiceMBean.class);
name = new ObjectName(StreamManagerMBean.OBJECT_NAME);
@ -1296,6 +1301,11 @@ public class NodeProbe implements AutoCloseable
return cmsProxy;
}
public AccordOperationsMBean getAccordOperationsProxy()
{
return accordProxy;
}
public GossiperMBean getGossProxy()
{
return gossProxy;

7
src/java/org/apache/cassandra/tools/NodeTool.java
View File

@ -278,6 +278,13 @@ public class NodeTool
.withCommands(ConsensusMigrationAdmin.ListCmd.class)
.withCommands(ConsensusMigrationAdmin.FinishMigration.class);
builder.withGroup("accord")
.withDescription("Manage the operation of Accord")
.withDefaultCommand(AccordAdmin.Describe.class)
.withCommand(AccordAdmin.Describe.class)
.withCommand(AccordAdmin.MarkStale.class)
.withCommand(AccordAdmin.MarkRejoining.class);
Cli<NodeToolCmdRunnable> parser = builder.build();
int status = 0;

69
src/java/org/apache/cassandra/tools/nodetool/AccordAdmin.java Normal file
View File

@ -0,0 +1,69 @@
/*
* 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.tools.nodetool;
import java.util.List;
import java.util.Map;
import io.airlift.airline.Arguments;
import io.airlift.airline.Command;
import org.apache.cassandra.tools.NodeProbe;
import org.apache.cassandra.tools.NodeTool;
public abstract class AccordAdmin extends NodeTool.NodeToolCmd
{
@Command(name = "describe", description = "Describe current cluster metadata relating to Accord")
public static class Describe extends NodeTool.NodeToolCmd
{
@Override
protected void execute(NodeProbe probe)
{
Map<String, String> info = probe.getAccordOperationsProxy().describe();
output.out.printf("Accord Service:%n");
output.out.printf("Epoch: %s%n", info.get("EPOCH"));
output.out.printf("Stale Replicas: %s%n", info.get("STALE_REPLICAS"));
}
}
@Command(name = "mark_stale", description = "Mark a replica as being stale and no longer able to participate in durability status coordination")
public static class MarkStale extends AccordAdmin
{
@Arguments(required = true, description = "One or more node IDs to mark stale", usage = "<nodeId>+")
public List<String> nodeIds;
@Override
protected void execute(NodeProbe probe)
{
probe.getAccordOperationsProxy().accordMarkStale(nodeIds);
}
}
@Command(name = "mark_rejoining", description = "Mark a stale replica as being allowed to participate in durability status coordination again")
public static class MarkRejoining extends AccordAdmin
{
@Arguments(required = true, description = "One or more node IDs to mark no longer stale", usage = "<nodeId>+")
public List<String> nodeIds;
@Override
protected void execute(NodeProbe probe)
{
probe.getAccordOperationsProxy().accordMarkRejoining(nodeIds);
}
}
}

2
test/distributed/org/apache/cassandra/distributed/mock/nodetool/InternalNodeProbe.java
View File

@ -46,6 +46,7 @@ import org.apache.cassandra.service.CacheServiceMBean;
import org.apache.cassandra.service.GCInspector;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.accord.AccordOperations;
import org.apache.cassandra.service.snapshot.SnapshotManager;
import org.apache.cassandra.streaming.StreamManager;
import org.apache.cassandra.tcm.CMSOperations;
@ -74,6 +75,7 @@ public class InternalNodeProbe extends NodeProbe
ssProxy = StorageService.instance;
snapshotProxy = SnapshotManager.instance;
cmsProxy = CMSOperations.instance;
accordProxy = AccordOperations.instance;
msProxy = MessagingService.instance();
streamProxy = StreamManager.instance;
compactionProxy = CompactionManager.instance;

99
test/distributed/org/apache/cassandra/distributed/test/accord/AccordNodetoolTest.java Normal file
View File

@ -0,0 +1,99 @@
/*
* 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.Collections;
import java.util.HashMap;
import java.util.Map;
import com.google.common.collect.ImmutableSet;
import org.junit.Test;
import accord.local.Node;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import static org.junit.Assert.assertEquals;
import static org.apache.cassandra.distributed.shared.ClusterUtils.getNodeId;
public class AccordNodetoolTest extends TestBaseImpl
{
@Test
public void testMarkSingleNode() throws Throwable
{
try (Cluster cluster = init(builder().withNodes(3).withConfig((config) -> config.with(Feature.NETWORK, Feature.GOSSIP)).start()))
{
cluster.get(1).nodetoolResult("accord", "mark_stale", "1").asserts().success();
cluster.get(1).runOnInstance(() -> assertEquals(ImmutableSet.of(new Node.Id(1)), ClusterMetadata.current().accordStaleReplicas.ids()));
cluster.get(1).nodetoolResult("accord", "describe").asserts().stdoutContains("Stale Replicas: 1");
// Reject the operation if the target node is already stale:
cluster.get(1).nodetoolResult("accord", "mark_stale", "1").asserts().failure().errorContains("it already is");
// Reject the operation if marking the node stale brings us below a quorum of non-stale nodes:
cluster.get(1).nodetoolResult("accord", "mark_stale", "2").asserts().failure().errorContains("that would leave fewer than a quorum");
// Reject the operation if the target node doesn't exist:
cluster.get(1).nodetoolResult("accord", "mark_stale", "4").asserts().failure().errorContains("not present in the directory");
cluster.get(1).nodetoolResult("accord", "mark_rejoining", "1").asserts().success();
cluster.get(1).runOnInstance(() -> assertEquals(Collections.emptySet(), ClusterMetadata.current().accordStaleReplicas.ids()));
cluster.get(1).nodetoolResult("accord", "mark_rejoining", "1").asserts().failure().errorContains("it is not stale");
cluster.get(1).nodetoolResult("accord", "mark_rejoining", "4").asserts().failure().errorContains("not present in the directory");
}
}
@Test
public void testMarkMultipleNodes() throws Throwable
{
try (Cluster cluster = init(builder().withNodes(5).withConfig((config) -> config.with(Feature.NETWORK, Feature.GOSSIP)).start()))
{
// Reject the operation if marking the node stale brings us below a quorum of non-stale nodes:
cluster.get(1).nodetoolResult("accord", "mark_stale", "1", "2", "3").asserts().failure().errorContains("that would leave fewer than a quorum");
cluster.get(1).nodetoolResult("accord", "mark_stale", "1", "2").asserts().success();
cluster.get(1).runOnInstance(() -> assertEquals(ImmutableSet.of(new Node.Id(1), new Node.Id(2)), ClusterMetadata.current().accordStaleReplicas.ids()));
cluster.get(1).nodetoolResult("accord", "describe").asserts().stdoutContains("Stale Replicas: 1,2");
// Reject the operation if a target node is already stale:
cluster.get(1).nodetoolResult("accord", "mark_stale", "1", "2").asserts().failure().errorContains("it already is");
// Reject the operation if a target node doesn't exist:
cluster.get(1).nodetoolResult("accord", "mark_stale", "4", "6").asserts().failure().errorContains("not present in the directory");
Map<Integer, Integer> nodeIdToNode = new HashMap<>();
for (int i = 1; i <= 5; i++)
nodeIdToNode.put(getNodeId(cluster.get(i)).id(), i);
// Remove the second stale node, and ensure the set of stale replicas is updated:
cluster.get(nodeIdToNode.get(2)).shutdown().get();
cluster.get(1).nodetoolResult("removenode", "2", "--force").asserts().success();
cluster.get(1).nodetoolResult("cms", "unregister", "2").asserts().success();
cluster.get(1).runOnInstance(() -> assertEquals(ImmutableSet.of(new Node.Id(1)), ClusterMetadata.current().accordStaleReplicas.ids()));
cluster.get(1).nodetoolResult("accord", "mark_rejoining", "1", "3").asserts().failure().errorContains("it is not stale");
cluster.get(1).nodetoolResult("accord", "mark_rejoining", "1", "6").asserts().failure().errorContains("not present in the directory");
cluster.get(1).runOnInstance(() -> assertEquals(ImmutableSet.of(new Node.Id(1)), ClusterMetadata.current().accordStaleReplicas.ids()));
}
}
}

10
test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java
View File

@ -58,6 +58,7 @@ import org.apache.cassandra.schema.Schema;
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.tcm.AtomicLongBackedProcessor;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
@ -158,7 +159,8 @@ public class ClusterMetadataTestHelper
LockedRanges.EMPTY,
InProgressSequences.EMPTY,
null,
ImmutableMap.of());
ImmutableMap.of(),
AccordStaleReplicas.EMPTY);
}
public static ClusterMetadata minimalForTesting(IPartitioner partitioner)
@ -173,7 +175,8 @@ public class ClusterMetadataTestHelper
null,
null,
null,
ImmutableMap.of());
ImmutableMap.of(),
AccordStaleReplicas.EMPTY);
}
public static ClusterMetadata minimalForTesting(Keyspaces keyspaces)
@ -188,7 +191,8 @@ public class ClusterMetadataTestHelper
null,
null,
null,
ImmutableMap.of());
ImmutableMap.of(),
AccordStaleReplicas.EMPTY);
}
public static ClusterMetadataService syncInstanceForTest()

10
test/unit/org/apache/cassandra/locator/MetaStrategyTest.java
View File

@ -26,14 +26,15 @@ import java.util.Map;
import java.util.Set;
import com.google.common.collect.ImmutableMap;
import org.apache.cassandra.service.accord.AccordFastPath;
import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState;
import org.junit.Assert;
import org.junit.Test;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.schema.DistributedSchema;
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.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.membership.Directory;
@ -93,7 +94,8 @@ public class MetaStrategyTest
LockedRanges.EMPTY,
InProgressSequences.EMPTY,
ConsensusMigrationState.EMPTY,
ImmutableMap.of());
ImmutableMap.of(),
AccordStaleReplicas.EMPTY);
}
@Test
@ -159,4 +161,4 @@ public class MetaStrategyTest
{
return new Location(dc, rack);
}
}
}

17
test/unit/org/apache/cassandra/service/accord/AccordFastPathCoordinatorTest.java
View File

@ -18,6 +18,15 @@
package org.apache.cassandra.service.accord;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import accord.local.Node;
import accord.topology.Shard;
import accord.topology.Topology;
@ -28,14 +37,6 @@ import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.schema.*;
import org.apache.cassandra.service.accord.AccordFastPath.Status;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import static org.apache.cassandra.service.accord.AccordTestUtils.*;

54
test/unit/org/apache/cassandra/service/accord/AccordStaleReplicasTest.java Normal file
View File

@ -0,0 +1,54 @@
/*
* 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.service.accord;
import java.util.HashSet;
import java.util.Set;
import org.junit.Test;
import accord.local.Node;
import accord.utils.AccordGens;
import accord.utils.Gen;
import accord.utils.Gens;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializers;
import org.apache.cassandra.tcm.serialization.Version;
import static accord.utils.Property.qt;
public class AccordStaleReplicasTest
{
@Test
public void serde()
{
try (DataOutputBuffer buffer = new DataOutputBuffer())
{
Gen<Set<Node.Id>> nodesGen = Gens.lists(AccordGens.nodes()).unique().ofSizeBetween(0, 9).map(nodes -> new HashSet<>(nodes));
Gen<Epoch> epochGen = AccordGens.epochs().map(Epoch::create);
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);
});
}
}
}

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

@ -301,6 +301,8 @@ public class ClusterMetadataTransformationTest
return metadata.accordFastPath;
else if (key == CONSENSUS_MIGRATION_STATE)
return metadata.consensusMigrationState;
else if (key == ACCORD_STALE_REPLICAS)
return metadata.accordStaleReplicas;
throw new IllegalArgumentException("Unknown metadata key " + key);
}

42
test/unit/org/apache/cassandra/tcm/serialization/AsymmetricMetadataSerializers.java Normal file
View File

@ -0,0 +1,42 @@
/*
* 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.serialization;
import java.io.IOException;
import org.assertj.core.api.Assertions;
import accord.utils.LazyToString;
import accord.utils.ReflectionUtils;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
public class AsymmetricMetadataSerializers
{
public static <In, Out> void testSerde(DataOutputBuffer output, AsymmetricMetadataSerializer<In, Out> serializer, In input, Version version) throws IOException
{
output.clear();
long expectedSize = serializer.serializedSize(input, version);
serializer.serialize(input, 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);
Out read = serializer.deserialize(in, version);
Assertions.assertThat(read).describedAs("The deserialized output does not match the serialized input; difference %s", new LazyToString(() -> ReflectionUtils.recursiveEquals(read, input).toString())).isEqualTo(input);
}
}

56
test/unit/org/apache/cassandra/tcm/transformations/AccordMarkRejoiningTest.java Normal file
View File

@ -0,0 +1,56 @@
/*
* 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.Collections;
import com.google.common.collect.ImmutableSet;
import org.junit.Test;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializers;
import org.apache.cassandra.tcm.serialization.Version;
public class AccordMarkRejoiningTest
{
@Test
public void shouldSerializeEmpty() throws IOException
{
DataOutputBuffer buffer = new DataOutputBuffer();
AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkRejoining.serializer, new AccordMarkRejoining(Collections.emptySet()), Version.V2);
}
@Test
public void shouldSerializeSingleton() throws IOException
{
DataOutputBuffer buffer = new DataOutputBuffer();
AccordMarkRejoining markStale = new AccordMarkRejoining(Collections.singleton(NodeId.fromString("1")));
AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkRejoining.serializer, markStale, Version.V2);
}
@Test
public void shouldSerializeMulti() throws IOException
{
DataOutputBuffer buffer = new DataOutputBuffer();
AccordMarkRejoining markStale = new AccordMarkRejoining(ImmutableSet.of(NodeId.fromString("1"), NodeId.fromString("2")));
AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkRejoining.serializer, markStale, Version.V2);
}
}

56
test/unit/org/apache/cassandra/tcm/transformations/AccordMarkStaleTest.java Normal file
View File

@ -0,0 +1,56 @@
/*
* 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.Collections;
import com.google.common.collect.ImmutableSet;
import org.junit.Test;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializers;
import org.apache.cassandra.tcm.serialization.Version;
public class AccordMarkStaleTest
{
@Test
public void shouldSerializeEmpty() throws IOException
{
DataOutputBuffer buffer = new DataOutputBuffer();
AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkStale.serializer, new AccordMarkStale(Collections.emptySet()), Version.V2);
}
@Test
public void shouldSerializeSingleton() throws IOException
{
DataOutputBuffer buffer = new DataOutputBuffer();
AccordMarkStale markStale = new AccordMarkStale(Collections.singleton(NodeId.fromString("1")));
AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkStale.serializer, markStale, Version.V2);
}
@Test
public void shouldSerializeMulti() throws IOException
{
DataOutputBuffer buffer = new DataOutputBuffer();
AccordMarkStale markStale = new AccordMarkStale(ImmutableSet.of(NodeId.fromString("1"), NodeId.fromString("2")));
AsymmetricMetadataSerializers.testSerde(buffer, AccordMarkStale.serializer, markStale, Version.V2);
}
}