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

[CASSANDRA-20736] Update ClusterMetadata.isCMSMember() to check using local node id

This commit is contained in:
Sam Tunnicliffe 2026-02-27 11:15:48 +00:00
parent 0acb57f38b
commit 46d109ff61
13 changed files with 25 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/java/org/apache/cassandra/metrics/TCMMetrics.java
View File

@ -29,7 +29,6 @@ import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.EpochAwareDebounce;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
import static org.apache.cassandra.tcm.transformations.cms.PrepareCMSReconfiguration.needsReconfiguration;
@ -94,7 +93,7 @@ public class TCMMetrics
isCMSMember = Metrics.register(factory.createMetricName("IsCMSMember"), () -> {
ClusterMetadata metadata = ClusterMetadata.currentNullable();
return metadata != null && metadata.isCMSMember(FBUtilities.getBroadcastAddressAndPort()) ? 1 : 0;
return metadata != null && metadata.isCMSMember() ? 1 : 0;
});
needsCMSReconfiguration = Metrics.register(factory.createMetricName("NeedsCMSReconfiguration"), () -> {

3
src/java/org/apache/cassandra/net/ResponseVerbHandler.java
View File

@ -30,7 +30,6 @@ import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.FBUtilities;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.cassandra.exceptions.RequestFailureReason.COORDINATOR_BEHIND;
@ -98,7 +97,7 @@ class ResponseVerbHandler implements IVerbHandler
if (SKIP_CATCHUP_FOR.contains(message.verb()))
return;
if (metadata.isCMSMember(FBUtilities.getBroadcastAddressAndPort()) && CMS_SKIP_CATCHUP_FOR.contains(message.verb()))
if (metadata.isCMSMember() && CMS_SKIP_CATCHUP_FOR.contains(message.verb()))
return;
// Gossip stage is single-threaded, so we may end up in a deadlock with after-commit hook

3
src/java/org/apache/cassandra/repair/RepairCoordinator.java
View File

@ -81,7 +81,6 @@ import org.apache.cassandra.tracing.TraceState;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.transport.Dispatcher;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.Throwables;
import org.apache.cassandra.utils.TimeUUID;
@ -402,7 +401,7 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
//calculation multiple times
Iterable<Range<Token>> keyspaceLocalRanges = getLocalReplicas.apply(state.keyspace).ranges();
boolean isMeta = Keyspace.open(state.keyspace).getMetadata().params.replication.isMeta();
boolean isCMS = ClusterMetadata.current().isCMSMember(FBUtilities.getBroadcastAddressAndPort());
boolean isCMS = ClusterMetadata.current().isCMSMember();
for (Range<Token> range : state.options.getRanges())
{
EndpointsForRange allForRange = ctx.repair().getNeighbors(state.keyspace, keyspaceLocalRanges, range);

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

@ -48,7 +48,6 @@ import org.apache.cassandra.tcm.sequences.ReconfigureCMS;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.tcm.transformations.Unregister;
import org.apache.cassandra.tcm.transformations.cms.AdvanceCMSReconfiguration;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MBeanWrapper;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
@ -217,7 +216,7 @@ public class CMSOperations implements CMSOperationsMBean
String members = metadata.fullCMSMembers().stream().sorted().map(Object::toString).collect(Collectors.joining(","));
info.put(MEMBERS, members);
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(IS_MEMBER, Boolean.toString(metadata.isCMSMember()));
info.put(SERVICE_STATE, ClusterMetadataService.state(metadata).toString());
info.put(IS_MIGRATING, Boolean.toString(cms.isMigrating()));
info.put(EPOCH, Long.toString(metadata.epoch.getEpoch()));

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

@ -237,13 +237,20 @@ public class ClusterMetadata
return cmsMembership.fullMembers();
}
public boolean isCMSMember()
{
if (epoch.isEqualOrBefore(Epoch.FIRST))
return isCMSMember(FBUtilities.getBroadcastAddressAndPort());
return fullCMSMemberIds().contains(localNodeId);
}
public boolean isCMSMember(InetAddressAndPort endpoint)
{
if (epoch.isAfter(Epoch.FIRST))
return fullCMSMembers().contains(endpoint);
if (epoch.isEqualOrBefore(Epoch.FIRST))
return cmsDataPlacement.reads.byEndpoint().keySet().contains(endpoint);
// special case to handle initialization of the CMS for the first time
return cmsDataPlacement.reads.byEndpoint().keySet().contains(endpoint);
return fullCMSMembers().contains(endpoint);
}
public Set<InetAddressAndPort> fullCMSMembers()

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

@ -176,7 +176,7 @@ public class ClusterMetadataService
// The node is a full member of the CMS if it has started participating in reads for distributed metadata table (which
// implies it is a write replica as well). In other words, it's a fully joined member of the replica set responsible for
// the distributed metadata table.
if (metadata.epoch.isEqualOrBefore(Epoch.FIRST) || ClusterMetadata.current().isCMSMember(FBUtilities.getBroadcastAddressAndPort()))
if (metadata.epoch.isEqualOrBefore(Epoch.FIRST) || metadata.isCMSMember())
return LOCAL;
return REMOTE;
@ -1103,6 +1103,6 @@ public class ClusterMetadataService
public enum State
{
LOCAL, REMOTE, GOSSIP, RESET
LOCAL, REMOTE, GOSSIP, RESET, OFFLINE_TOOL
}
}

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

@ -66,7 +66,7 @@ public class PaxosBackedProcessor extends AbstractLocalProcessor
@Override
protected boolean acceptCommit(ClusterMetadata metadata)
{
return metadata.isCMSMember(FBUtilities.getBroadcastAddressAndPort());
return metadata.isCMSMember();
}
@Override

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

@ -402,7 +402,7 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
DatabaseDescriptor.getPartitioner().getClass().getCanonicalName(),
metadata.partitioner.getClass().getCanonicalName()));
if (!metadata.isCMSMember(FBUtilities.getBroadcastAddressAndPort()))
if (!metadata.isCMSMember())
throw new IllegalStateException("When reinitializing with cluster metadata, we must be in the CMS");
metadata = metadata.forceEpoch(metadata.epoch.nextEpoch());

2
src/java/org/apache/cassandra/tools/CMSOfflineTool.java
View File

@ -609,7 +609,7 @@ public class CMSOfflineTool implements Runnable
output.out.printf("Cluster Metadata Service:%n");
output.out.printf("Members: %s%n", members);
output.out.printf("Needs reconfiguration: %s%n", needsReconfiguration(metadata));
output.out.printf("Service State: %s%n", ClusterMetadataService.state(metadata));
output.out.printf("Service State: %s%n", ClusterMetadataService.State.OFFLINE_TOOL);
output.out.printf("Epoch: %s%n", metadata.epoch.getEpoch());
output.out.printf("Replication factor: %s%n", ReplicationParams.meta(metadata).toString());
}

5
test/distributed/org/apache/cassandra/distributed/test/hostreplacement/AssassinateCMSNodeTest.java
View File

@ -27,7 +27,6 @@ import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
@ -46,8 +45,8 @@ public class AssassinateCMSNodeTest extends TestBaseImpl
InetSocketAddress toAssassinate = cluster.get(2).broadcastAddress();
cluster.get(2).shutdown().get();
cluster.get(1).nodetoolResult("assassinate", toAssassinate.getHostString()).asserts().success();
cluster.get(1).runOnInstance(() -> assertTrue(ClusterMetadata.current().isCMSMember(FBUtilities.getBroadcastAddressAndPort())));
cluster.get(3).runOnInstance(() -> assertTrue(ClusterMetadata.current().isCMSMember(FBUtilities.getBroadcastAddressAndPort())));
cluster.get(1).runOnInstance(() -> assertTrue(ClusterMetadata.current().isCMSMember()));
cluster.get(3).runOnInstance(() -> assertTrue(ClusterMetadata.current().isCMSMember()));
cluster.get(1).nodetoolResult("cms").asserts().success();
}
}

2
test/distributed/org/apache/cassandra/distributed/test/log/ReconfigureCMSTest.java
View File

@ -216,7 +216,7 @@ public class ReconfigureCMSTest extends FuzzTestBase
awaitRingJoin(replacingNode, cluster.get(1));
replacingNode.runOnInstance(() -> {
ClusterMetadata metadata = ClusterMetadata.current();
assertTrue(metadata.isCMSMember(FBUtilities.getBroadcastAddressAndPort()));
assertTrue(metadata.isCMSMember());
assertEquals(3, metadata.fullCMSMembers().size());
});
}

3
test/distributed/org/apache/cassandra/distributed/test/tcm/CMSPlacementAfterReplacementTest.java
View File

@ -34,7 +34,6 @@ import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.distributed.shared.ClusterUtils.addInstance;
import static org.apache.cassandra.distributed.shared.ClusterUtils.awaitRingJoin;
@ -80,7 +79,7 @@ public class CMSPlacementAfterReplacementTest extends TestBaseImpl
IInvokableInstance nodeToRemove = cluster.get(2);
cluster.get(1).nodetoolResult("cms", "reconfigure", "3").asserts().success();
cluster.get(2).runOnInstance(() -> {
assertTrue(ClusterMetadata.current().isCMSMember(FBUtilities.getBroadcastAddressAndPort()));
assertTrue(ClusterMetadata.current().isCMSMember());
});
nodeToRemove.shutdown().get();
IInvokableInstance replacingNode = addInstance(cluster, nodeToRemove.config(),

2
test/unit/org/apache/cassandra/tools/CMSOfflineToolTest.java
View File

@ -670,7 +670,7 @@ public class CMSOfflineToolTest extends OfflineToolUtils
"Cluster Metadata Service:\n" +
"Members: /127.0.0.1:" + storagePort + ",/127.0.0.2:" + storagePort + ",/127.0.0.3:" + storagePort + '\n' +
"Needs reconfiguration: false\n" +
"Service State: LOCAL\n" +
"Service State: " + ClusterMetadataService.State.OFFLINE_TOOL + '\n' +
"Epoch: 2\n" +
"Replication factor: ReplicationParams{class=org.apache.cassandra.locator.MetaStrategy, datacenter1=3}\n";
assertThat(result.getStdout()).isEqualTo(expectedOutput);