[PR] Make sure system.peers and peers_v2 match cluster metadata [cassandra]

["minal-kyada (via GitHub)" <gi...@apache.org>]

minal-kyada opened a new pull request, #4630:
URL: https://github.com/apache/cassandra/pull/4630

   **Description:**                                                                    
   system.peers and system.peers_v2 can drift out of sync with ClusterMetadata, causing older clients and tools that read these legacy tables to observe incorrect cluster topology.
   
   Adds SystemPeersValidator which reconciles both peer tables against ClusterMetadata on startup: removing stale entries for nodes no longer in the cluster and repairing missing entries for joined nodes. Also exposes this as a JMX operation via StorageServiceMBean so operators can trigger it on demand without restarting.
   
   patch by @minal-kyada; reviewed by @krummas for [CASSANDRA-21187](https://issues.apache.org/jira/browse/CASSANDRA-21187)
   
   Co-authored-by: Name1 <email1>
   Co-authored-by: Name2 <email2>
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] Make sure system.peers and peers_v2 match cluster metadata [cassandra]

["minal-kyada (via GitHub)" <gi...@apache.org>]

minal-kyada commented on code in PR #4630:
URL: https://github.com/apache/cassandra/pull/4630#discussion_r2897648505


##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.UTF8Type;
+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.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata on startup.
+ * This is critical for backward compatibility as older clients and tools read from these
+ * legacy tables while TCM uses ClusterMetadata as the source of truth.
+ *
+ * The validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ */
+public class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        Map<InetAddressAndPort, UntypedResultSet.Row> peersV2Rows = getPeersV2Rows();
+        Map<InetAddress, UntypedResultSet.Row> legacyPeersRows = getLegacyPeersRows();
+
+        Map<InetAddressAndPort, NodeId> expectedEndpoints = new HashMap<>();
+        Map<InetAddress, NodeId> expectedAddresses = new HashMap<>();
+        for (NodeId nodeId : getExpectedPeerNodes(metadata))

Review Comment:
   I had this because I wanted to eliminate localEndpoints, but there's a better way to do it by just having allJoinedEndpoints and skipping the local endpoint. Will change it! Thanks Sam!



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] Make sure system.peers and peers_v2 match cluster metadata [cassandra]

["ifesdjeen (via GitHub)" <gi...@apache.org>]

ifesdjeen commented on code in PR #4630:
URL: https://github.com/apache/cassandra/pull/4630#discussion_r3001832731


##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.UTF8Type;
+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.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata.
+ * These tables are maintained for existing clients and tools which read from them to determine
+ * topology and schema information, while Cassandra itself uses ClusterMetadata as the source of truth.
+ *
+ * This validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ *
+ * The system_views.peers virtual table provides a live view on the current ClusterMetadata
+ * and includes all members of the cluster, unlike the legacy tables which exclude the local node.
+ */
+public final class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+    private static final String SELECT_PEERS_V2_QUERY = String.format("SELECT * FROM %s.%s",
+                                                                      SchemaConstants.SYSTEM_KEYSPACE_NAME, PEERS_V2);
+    private static final String SELECT_PEERS_QUERY = String.format("SELECT * FROM %s.%s",
+                                                                    SchemaConstants.SYSTEM_KEYSPACE_NAME, LEGACY_PEERS);
+    private static final String DELETE_PEERS_V2_QUERY = String.format("DELETE FROM %s.%s WHERE peer = ? AND peer_port = ?",
+                                                                     SchemaConstants.SYSTEM_KEYSPACE_NAME, PEERS_V2);
+    private static final String DELETE_PEERS_QUERY = String.format("DELETE FROM %s.%s WHERE peer = ?",
+                                                                   SchemaConstants.SYSTEM_KEYSPACE_NAME, LEGACY_PEERS);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        Map<InetAddressAndPort, UntypedResultSet.Row> peersV2Rows = getPeersV2Rows();
+        Map<InetAddress, UntypedResultSet.Row> peersRows = getPeersRows();
+
+        Map<InetAddressAndPort, NodeId> knownEndpoints = new HashMap<>();
+        Set<InetAddress> knownAddresses = new HashSet<>();
+        InetAddressAndPort localEndpoint = FBUtilities.getBroadcastAddressAndPort();
+        for (InetAddressAndPort endpoint : metadata.directory.allJoinedEndpoints())
+        {
+            if (endpoint.equals(localEndpoint))
+                continue;
+            knownEndpoints.put(endpoint, metadata.directory.peerId(endpoint));
+            knownAddresses.add(endpoint.getAddress());
+        }
+
+        for (InetAddressAndPort endpoint : peersV2Rows.keySet())
+        {
+            if (!knownEndpoints.containsKey(endpoint))
+            {
+                logger.info("Removing stale peer {} from {}", endpoint, PEERS_V2);
+                executeInternal(DELETE_PEERS_V2_QUERY, endpoint.getAddress(), endpoint.getPort());
+            }
+        }
+
+        for (InetAddress address : peersRows.keySet())
+        {
+            if (!knownAddresses.contains(address))
+            {
+                logger.info("Removing stale peer {} from {}", address, LEGACY_PEERS);
+                executeInternal(DELETE_PEERS_QUERY, address);
+            }
+        }
+
+        for (Map.Entry<InetAddressAndPort, NodeId> entry : knownEndpoints.entrySet())
+        {
+            NodeId nodeId = entry.getValue();
+            InetAddressAndPort endpoint = entry.getKey();
+            UntypedResultSet.Row peersV2Row = peersV2Rows.get(endpoint);
+            UntypedResultSet.Row peersRow = peersRows.get(endpoint.getAddress());
+
+            boolean peersV2NeedsRepair = needsRepair(peersV2Row, endpoint, PEERS_V2, nodeId, metadata);
+            boolean peersNeedsRepair = needsRepair(peersRow, endpoint, LEGACY_PEERS, nodeId, metadata);
+
+            if (peersV2NeedsRepair || peersNeedsRepair)
+                PeersTable.updateLegacyPeerTable(nodeId, metadata, metadata);
+        }
+    }
+
+    private static boolean needsRepair(UntypedResultSet.Row row, InetAddressAndPort endpoint,
+                                       String table, NodeId nodeId, ClusterMetadata metadata)
+    {
+        if (row == null)
+        {
+            logger.info("Adding missing peer {} to {}", endpoint, table);
+            return true;
+        }
+        boolean isEquivalent = PEERS_V2.equals(table)
+                               ? peersV2RowIsEquivalent(row, nodeId, metadata)
+                               : peersRowIsEquivalent(row, nodeId, metadata);
+        if (!isEquivalent)
+        {
+            logger.info("Repairing mismatched peer {} in {}: {}", endpoint, table, row);
+            return true;
+        }
+        return false;
+    }
+
+    private static boolean peersV2RowIsEquivalent(UntypedResultSet.Row row, NodeId nodeId, ClusterMetadata metadata)
+    {
+        NodeAddresses addresses = metadata.directory.getNodeAddresses(nodeId);
+        return commonColumnsAreEquivalent(row, nodeId, addresses, metadata)
+               && row.has("preferred_port") && row.getInt("preferred_port") == addresses.broadcastAddress.getPort()
+               && row.has("native_port") && row.getInt("native_port") == addresses.nativeAddress.getPort();
+    }
+
+    private static boolean peersRowIsEquivalent(UntypedResultSet.Row row, NodeId nodeId, ClusterMetadata metadata)
+    {
+        NodeAddresses addresses = metadata.directory.getNodeAddresses(nodeId);
+        return commonColumnsAreEquivalent(row, nodeId, addresses, metadata);
+    }
+
+    private static boolean commonColumnsAreEquivalent(UntypedResultSet.Row row, NodeId nodeId,
+                                                      NodeAddresses addresses, ClusterMetadata metadata)
+    {
+        Location location = metadata.directory.location(nodeId);
+        // This column is differently named in the peers and peers_v2 tables
+        String nativeAddressColumn = row.has("native_address") ? "native_address" : "rpc_address";

Review Comment:
   should we just pass from outside which table we're dealing with and assert correspondingly?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] Make sure system.peers and peers_v2 match cluster metadata [cassandra]

["beobal (via GitHub)" <gi...@apache.org>]

beobal commented on code in PR #4630:
URL: https://github.com/apache/cassandra/pull/4630#discussion_r2885410112


##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.UTF8Type;
+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.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**

Review Comment:
   I have a couple of suggestions for this comment: 
   * I would soften the language slightly
   * Remove mention of startup, as it can also be triggered directly
   * Mention that there is a virtual table which presents a live view of ClusterMetadata.
   * I prefer not to use TCM in the codebase as it's a bit vague and unspecific and sometimes is incorrectly used as a kind of catch-all label for anything related to internal metadata.
    
    So, I'd reword this slightly to:
   ```
   Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata.
   These tables are maintained for existing clients and tools which read from them to determine 
   topology and schema information, while Cassandra itself uses ClusterMetadata as the source of truth.
   
   This validator detects inconsistencies and automatically repairs them by synchronizing
   the peers tables with the current ClusterMetadata.
   
   The system_views.peers virtual table provides a live view on the current ClusterMetadata 
   and includes all members of the cluster, unlike the legacy tables which exclude the local node. 
   ```
   



##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.UTF8Type;
+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.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata on startup.
+ * This is critical for backward compatibility as older clients and tools read from these
+ * legacy tables while TCM uses ClusterMetadata as the source of truth.
+ *
+ * The validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ */
+public class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        Map<InetAddressAndPort, UntypedResultSet.Row> peersV2Rows = getPeersV2Rows();
+        Map<InetAddress, UntypedResultSet.Row> legacyPeersRows = getLegacyPeersRows();
+
+        Map<InetAddressAndPort, NodeId> expectedEndpoints = new HashMap<>();
+        Map<InetAddress, NodeId> expectedAddresses = new HashMap<>();
+        for (NodeId nodeId : getExpectedPeerNodes(metadata))
+        {
+            InetAddressAndPort endpoint = metadata.directory.endpoint(nodeId);
+            expectedEndpoints.put(endpoint, nodeId);
+            expectedAddresses.put(endpoint.getAddress(), nodeId);
+        }
+
+        String deleteV2Query = String.format("DELETE FROM %s.%s WHERE peer = ? AND peer_port = ?",

Review Comment:
   This can be static and final



##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.UTF8Type;
+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.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata on startup.
+ * This is critical for backward compatibility as older clients and tools read from these
+ * legacy tables while TCM uses ClusterMetadata as the source of truth.
+ *
+ * The validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ */
+public class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        Map<InetAddressAndPort, UntypedResultSet.Row> peersV2Rows = getPeersV2Rows();
+        Map<InetAddress, UntypedResultSet.Row> legacyPeersRows = getLegacyPeersRows();
+
+        Map<InetAddressAndPort, NodeId> expectedEndpoints = new HashMap<>();
+        Map<InetAddress, NodeId> expectedAddresses = new HashMap<>();
+        for (NodeId nodeId : getExpectedPeerNodes(metadata))
+        {
+            InetAddressAndPort endpoint = metadata.directory.endpoint(nodeId);
+            expectedEndpoints.put(endpoint, nodeId);
+            expectedAddresses.put(endpoint.getAddress(), nodeId);
+        }
+
+        String deleteV2Query = String.format("DELETE FROM %s.%s WHERE peer = ? AND peer_port = ?",
+                                               SchemaConstants.SYSTEM_KEYSPACE_NAME, PEERS_V2);
+        for (InetAddressAndPort endpoint : peersV2Rows.keySet())
+        {
+            if (!expectedEndpoints.containsKey(endpoint))
+            {
+                logger.info("Removing stale peer {} from {}", endpoint, PEERS_V2);
+                executeInternal(deleteV2Query, endpoint.getAddress(), endpoint.getPort());
+            }
+        }
+
+        String deleteLegacyQuery = String.format("DELETE FROM %s.%s WHERE peer = ?",

Review Comment:
   This can be static and final



##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.UTF8Type;
+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.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata on startup.
+ * This is critical for backward compatibility as older clients and tools read from these
+ * legacy tables while TCM uses ClusterMetadata as the source of truth.
+ *
+ * The validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ */
+public class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        Map<InetAddressAndPort, UntypedResultSet.Row> peersV2Rows = getPeersV2Rows();
+        Map<InetAddress, UntypedResultSet.Row> legacyPeersRows = getLegacyPeersRows();
+
+        Map<InetAddressAndPort, NodeId> expectedEndpoints = new HashMap<>();
+        Map<InetAddress, NodeId> expectedAddresses = new HashMap<>();
+        for (NodeId nodeId : getExpectedPeerNodes(metadata))

Review Comment:
   You could do this without `getExpectedPeerNodes` by just iterating `metadata.directory.allJoinedEndpoints()` here.



##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.UTF8Type;
+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.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata on startup.
+ * This is critical for backward compatibility as older clients and tools read from these
+ * legacy tables while TCM uses ClusterMetadata as the source of truth.
+ *
+ * The validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ */
+public class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        Map<InetAddressAndPort, UntypedResultSet.Row> peersV2Rows = getPeersV2Rows();
+        Map<InetAddress, UntypedResultSet.Row> legacyPeersRows = getLegacyPeersRows();
+
+        Map<InetAddressAndPort, NodeId> expectedEndpoints = new HashMap<>();
+        Map<InetAddress, NodeId> expectedAddresses = new HashMap<>();

Review Comment:
   This only needs to be a `Set`, the node id is never retrieved using the address.



##########
test/unit/org/apache/cassandra/db/SystemPeersValidatorTest.java:
##########
@@ -0,0 +1,265 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.HashSet;
+import java.util.UUID;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.ServerTestUtils;
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.distributed.test.log.CMSTestBase;
+import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
+import org.apache.cassandra.harry.model.TokenPlacementModel;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.schema.SchemaConstants;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.tcm.AtomicLongBackedProcessor;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.ClusterMetadataService;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class SystemPeersValidatorTest
+{
+    private CMSTestBase.CMSSut sut;
+    private InetAddressAndPort peerEndpoint;
+
+    @BeforeClass
+    public static void beforeClass()
+    {
+        ServerTestUtils.prepareServerNoRegister();
+    }
+
+    @Before
+    public void before() throws Exception
+    {
+        ClusterMetadataService.unsetInstance();
+        sut = new CMSTestBase.CMSSut(AtomicLongBackedProcessor::new, false,
+                                     new TokenPlacementModel.SimpleReplicationFactor(3));
+        ClusterMetadataTestHelper.register(2);
+        ClusterMetadataTestHelper.join(2, 2);
+        peerEndpoint = ClusterMetadata.current().directory.endpoint(ClusterMetadataTestHelper.nodeId(2));
+        cleanupPeersTables();
+    }
+
+    @After
+    public void after()
+    {
+        if (sut != null)
+            sut.close();
+    }
+
+    @Test
+    public void testNoRepairWhenTablesAreConsistent()
+    {
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertEquals("peers_v2 should have 1 peer", 1, countEntries(PEERS_V2));
+        assertEquals("legacy peers should have 1 peer", 1, countEntries(LEGACY_PEERS));
+
+        // Second call should be a no-op
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertEquals("peers_v2 count should not change on second call", 1, countEntries(PEERS_V2));
+        assertEquals("legacy peers count should not change on second call", 1, countEntries(LEGACY_PEERS));
+    }
+
+    @Test
+    public void testStaleEntriesAreRemovedFromBothTables() throws UnknownHostException
+    {
+        InetAddressAndPort staleEndpoint = InetAddressAndPort.getByName("127.0.0.99");
+        insertStalePeerEntry(staleEndpoint, PEERS_V2);
+        insertStalePeerEntry(staleEndpoint, LEGACY_PEERS);
+
+        assertTrue(entryExistsInPeersV2(staleEndpoint));
+        assertTrue(entryExistsInLegacyPeers(staleEndpoint.getAddress()));
+
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertFalse("Stale entry should be removed from peers_v2", entryExistsInPeersV2(staleEndpoint));
+        assertFalse("Stale entry should be removed from legacy peers", entryExistsInLegacyPeers(staleEndpoint.getAddress()));
+    }
+
+    @Test
+    public void testStaleEntryOnlyInPeersV2IsRemoved() throws UnknownHostException
+    {
+        InetAddressAndPort staleEndpoint = InetAddressAndPort.getByName("127.0.0.99");
+        insertStalePeerEntry(staleEndpoint, PEERS_V2);
+
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertFalse("Stale entry should be removed from peers_v2", entryExistsInPeersV2(staleEndpoint));
+    }
+
+    @Test
+    public void testStaleEntryOnlyInLegacyPeersIsRemoved() throws UnknownHostException
+    {
+        InetAddressAndPort staleEndpoint = InetAddressAndPort.getByName("127.0.0.99");
+        insertStalePeerEntry(staleEndpoint, LEGACY_PEERS);
+
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertFalse("Stale entry should be removed from legacy peers", entryExistsInLegacyPeers(staleEndpoint.getAddress()));
+    }
+
+    @Test
+    public void testMissingPeerIsAddedToBothTables()
+    {
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+        removePeerEntry(peerEndpoint);
+
+        assertFalse(entryExistsInPeersV2(peerEndpoint));
+        assertFalse(entryExistsInLegacyPeers(peerEndpoint.getAddress()));
+
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertTrue("Missing peer should be added to peers_v2", entryExistsInPeersV2(peerEndpoint));
+        assertTrue("Missing peer should be added to legacy peers", entryExistsInLegacyPeers(peerEndpoint.getAddress()));
+    }
+
+    @Test
+    public void testStaleFieldInPeersV2IsRepaired()
+    {
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        executeInternal(String.format("UPDATE %s.%s SET data_center = 'stale-dc' WHERE peer = ? AND peer_port = ?",
+                                      SchemaConstants.SYSTEM_KEYSPACE_NAME, PEERS_V2),
+                        peerEndpoint.getAddress(), peerEndpoint.getPort());
+
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        String expectedDc = ClusterMetadata.current().directory.location(ClusterMetadataTestHelper.nodeId(2)).datacenter;
+        assertEquals("data_center in peers_v2 should match ClusterMetadata",
+                     expectedDc, getDataCenter(peerEndpoint, PEERS_V2));
+    }
+
+    @Test
+    public void testStaleFieldInLegacyPeersIsRepaired()
+    {
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        executeInternal(String.format("UPDATE %s.%s SET data_center = 'stale-dc' WHERE peer = ?",
+                                      SchemaConstants.SYSTEM_KEYSPACE_NAME, LEGACY_PEERS),
+                        peerEndpoint.getAddress());
+
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        String expectedDc = ClusterMetadata.current().directory.location(ClusterMetadataTestHelper.nodeId(2)).datacenter;
+        assertEquals("data_center in legacy peers should match ClusterMetadata",
+                     expectedDc, getDataCenter(peerEndpoint, LEGACY_PEERS));
+    }
+
+    @Test
+    public void testJmxEntryPointRepairsMissingPeer()
+    {
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+        removePeerEntry(peerEndpoint);
+
+        StorageService.instance.validateAndRepairPeersMetadata();
+
+        assertTrue("JMX repair should restore peer in peers_v2", entryExistsInPeersV2(peerEndpoint));
+        assertTrue("JMX repair should restore peer in legacy peers", entryExistsInLegacyPeers(peerEndpoint.getAddress()));
+    }
+
+    private void cleanupPeersTables()
+    {
+        executeInternal(String.format("TRUNCATE %s.%s", SchemaConstants.SYSTEM_KEYSPACE_NAME, PEERS_V2));
+        executeInternal(String.format("TRUNCATE %s.%s", SchemaConstants.SYSTEM_KEYSPACE_NAME, LEGACY_PEERS));
+    }
+
+    private int countEntries(String table)
+    {
+        UntypedResultSet result = executeInternal(String.format("SELECT COUNT(*) FROM %s.%s",
+                                                                SchemaConstants.SYSTEM_KEYSPACE_NAME, table));
+        return (int) result.one().getLong("count");
+    }
+
+    private boolean entryExistsInPeersV2(InetAddressAndPort endpoint)
+    {
+        UntypedResultSet result = executeInternal(
+            String.format("SELECT peer FROM %s.%s WHERE peer = ? AND peer_port = ?",
+                          SchemaConstants.SYSTEM_KEYSPACE_NAME, PEERS_V2),
+            endpoint.getAddress(), endpoint.getPort());
+        return !result.isEmpty();
+    }
+
+    private boolean entryExistsInLegacyPeers(InetAddress address)
+    {
+        UntypedResultSet result = executeInternal(
+            String.format("SELECT peer FROM %s.%s WHERE peer = ?",
+                          SchemaConstants.SYSTEM_KEYSPACE_NAME, LEGACY_PEERS),
+            address);
+        return !result.isEmpty();
+    }
+
+    private String getDataCenter(InetAddressAndPort endpoint, String table)
+    {
+        UntypedResultSet result;
+        if (table.equals(PEERS_V2))
+            result = executeInternal(
+                String.format("SELECT data_center FROM %s.%s WHERE peer = ? AND peer_port = ?",
+                              SchemaConstants.SYSTEM_KEYSPACE_NAME, table),

Review Comment:
   I generally tend to add static imports for things like `SchemaConstants.SYSTEM_KEYSPACE_NAME` as it removes a fair bit of noise.



##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.UTF8Type;
+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.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata on startup.
+ * This is critical for backward compatibility as older clients and tools read from these
+ * legacy tables while TCM uses ClusterMetadata as the source of truth.
+ *
+ * The validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ */
+public class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        Map<InetAddressAndPort, UntypedResultSet.Row> peersV2Rows = getPeersV2Rows();
+        Map<InetAddress, UntypedResultSet.Row> legacyPeersRows = getLegacyPeersRows();

Review Comment:
   Nit: I would generally just refer to these two tables  as the `peers/peersV2` tables, so `peersRows/getPeersRows & peersV2Rows/getPeersV2Rows` here. Same for the CQL queries (i.e. `deletePeersQuery/deletePeersV2Query` and so on. 



##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.UTF8Type;
+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.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata on startup.
+ * This is critical for backward compatibility as older clients and tools read from these
+ * legacy tables while TCM uses ClusterMetadata as the source of truth.
+ *
+ * The validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ */
+public class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        Map<InetAddressAndPort, UntypedResultSet.Row> peersV2Rows = getPeersV2Rows();
+        Map<InetAddress, UntypedResultSet.Row> legacyPeersRows = getLegacyPeersRows();
+
+        Map<InetAddressAndPort, NodeId> expectedEndpoints = new HashMap<>();

Review Comment:
   nit: I would consider renaming to `knownEndpoints`



##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.UTF8Type;
+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.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata on startup.
+ * This is critical for backward compatibility as older clients and tools read from these
+ * legacy tables while TCM uses ClusterMetadata as the source of truth.
+ *
+ * The validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ */
+public class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        Map<InetAddressAndPort, UntypedResultSet.Row> peersV2Rows = getPeersV2Rows();
+        Map<InetAddress, UntypedResultSet.Row> legacyPeersRows = getLegacyPeersRows();
+
+        Map<InetAddressAndPort, NodeId> expectedEndpoints = new HashMap<>();
+        Map<InetAddress, NodeId> expectedAddresses = new HashMap<>();
+        for (NodeId nodeId : getExpectedPeerNodes(metadata))
+        {
+            InetAddressAndPort endpoint = metadata.directory.endpoint(nodeId);
+            expectedEndpoints.put(endpoint, nodeId);
+            expectedAddresses.put(endpoint.getAddress(), nodeId);
+        }
+
+        String deleteV2Query = String.format("DELETE FROM %s.%s WHERE peer = ? AND peer_port = ?",
+                                               SchemaConstants.SYSTEM_KEYSPACE_NAME, PEERS_V2);
+        for (InetAddressAndPort endpoint : peersV2Rows.keySet())
+        {
+            if (!expectedEndpoints.containsKey(endpoint))
+            {
+                logger.info("Removing stale peer {} from {}", endpoint, PEERS_V2);
+                executeInternal(deleteV2Query, endpoint.getAddress(), endpoint.getPort());
+            }
+        }
+
+        String deleteLegacyQuery = String.format("DELETE FROM %s.%s WHERE peer = ?",
+                                                   SchemaConstants.SYSTEM_KEYSPACE_NAME,
+                                                   LEGACY_PEERS);
+        for (InetAddress address : legacyPeersRows.keySet())
+        {
+            if (!expectedAddresses.containsKey(address))
+            {
+                logger.info("Removing stale peer {} from {}", address, LEGACY_PEERS);
+                executeInternal(deleteLegacyQuery, address);
+            }
+        }
+
+        for (Map.Entry<InetAddressAndPort, NodeId> entry : expectedEndpoints.entrySet())
+        {
+            NodeId nodeId = entry.getValue();
+            InetAddressAndPort endpoint = entry.getKey();
+            UntypedResultSet.Row v2Row = peersV2Rows.get(endpoint);
+            UntypedResultSet.Row legacyRow = legacyPeersRows.get(endpoint.getAddress());
+
+            List<String> v2Discrepancies = collectV2Discrepancies(v2Row, nodeId, metadata);

Review Comment:
   I think this can be trimmed down a bit, as there's no need to collect differences and log them individually. 
   
   I would implement a boolean method (or methods) which checks equivalence between a row from one of the tables and the values in cluster metadata. If not equivalent, then just log that there was a mismatch and include everything in the row (you can use `UntypedResultSet.Row::toString` for that as it handles all the data typing and so on for you).
   
   This hypothetical equivalence method should take into account potentially missing/unset columns,  which I don't think the existing checks do (i.e. what if the row has has no value for one of the columns?)
   
   There is a lot of commonality between the two table schemas so it should be possible to come up with something reasonably generic to check both (may need a couple of methods, but I expect they can share a lot).  



##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.UTF8Type;
+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.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata on startup.
+ * This is critical for backward compatibility as older clients and tools read from these
+ * legacy tables while TCM uses ClusterMetadata as the source of truth.
+ *
+ * The validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ */
+public class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        Map<InetAddressAndPort, UntypedResultSet.Row> peersV2Rows = getPeersV2Rows();
+        Map<InetAddress, UntypedResultSet.Row> legacyPeersRows = getLegacyPeersRows();
+
+        Map<InetAddressAndPort, NodeId> expectedEndpoints = new HashMap<>();
+        Map<InetAddress, NodeId> expectedAddresses = new HashMap<>();

Review Comment:
   nit: as well as changing to a `Set` I would maybe rename to `knownAddresses`



##########
test/unit/org/apache/cassandra/db/SystemPeersValidatorTest.java:
##########
@@ -0,0 +1,265 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.HashSet;
+import java.util.UUID;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.ServerTestUtils;
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.distributed.test.log.CMSTestBase;
+import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
+import org.apache.cassandra.harry.model.TokenPlacementModel;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.schema.SchemaConstants;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.tcm.AtomicLongBackedProcessor;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.ClusterMetadataService;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class SystemPeersValidatorTest
+{
+    private CMSTestBase.CMSSut sut;
+    private InetAddressAndPort peerEndpoint;
+
+    @BeforeClass
+    public static void beforeClass()
+    {
+        ServerTestUtils.prepareServerNoRegister();
+    }
+
+    @Before
+    public void before() throws Exception
+    {
+        ClusterMetadataService.unsetInstance();
+        sut = new CMSTestBase.CMSSut(AtomicLongBackedProcessor::new, false,
+                                     new TokenPlacementModel.SimpleReplicationFactor(3));
+        ClusterMetadataTestHelper.register(2);
+        ClusterMetadataTestHelper.join(2, 2);
+        peerEndpoint = ClusterMetadata.current().directory.endpoint(ClusterMetadataTestHelper.nodeId(2));
+        cleanupPeersTables();
+    }
+
+    @After
+    public void after()
+    {
+        if (sut != null)
+            sut.close();
+    }
+
+    @Test
+    public void testNoRepairWhenTablesAreConsistent()
+    {
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertEquals("peers_v2 should have 1 peer", 1, countEntries(PEERS_V2));
+        assertEquals("legacy peers should have 1 peer", 1, countEntries(LEGACY_PEERS));
+
+        // Second call should be a no-op
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertEquals("peers_v2 count should not change on second call", 1, countEntries(PEERS_V2));
+        assertEquals("legacy peers count should not change on second call", 1, countEntries(LEGACY_PEERS));
+    }
+
+    @Test
+    public void testStaleEntriesAreRemovedFromBothTables() throws UnknownHostException
+    {
+        InetAddressAndPort staleEndpoint = InetAddressAndPort.getByName("127.0.0.99");
+        insertStalePeerEntry(staleEndpoint, PEERS_V2);
+        insertStalePeerEntry(staleEndpoint, LEGACY_PEERS);
+
+        assertTrue(entryExistsInPeersV2(staleEndpoint));
+        assertTrue(entryExistsInLegacyPeers(staleEndpoint.getAddress()));
+
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertFalse("Stale entry should be removed from peers_v2", entryExistsInPeersV2(staleEndpoint));
+        assertFalse("Stale entry should be removed from legacy peers", entryExistsInLegacyPeers(staleEndpoint.getAddress()));
+    }
+
+    @Test
+    public void testStaleEntryOnlyInPeersV2IsRemoved() throws UnknownHostException
+    {
+        InetAddressAndPort staleEndpoint = InetAddressAndPort.getByName("127.0.0.99");
+        insertStalePeerEntry(staleEndpoint, PEERS_V2);
+
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertFalse("Stale entry should be removed from peers_v2", entryExistsInPeersV2(staleEndpoint));
+    }
+
+    @Test
+    public void testStaleEntryOnlyInLegacyPeersIsRemoved() throws UnknownHostException
+    {
+        InetAddressAndPort staleEndpoint = InetAddressAndPort.getByName("127.0.0.99");
+        insertStalePeerEntry(staleEndpoint, LEGACY_PEERS);
+
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertFalse("Stale entry should be removed from legacy peers", entryExistsInLegacyPeers(staleEndpoint.getAddress()));
+    }
+
+    @Test
+    public void testMissingPeerIsAddedToBothTables()
+    {
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+        removePeerEntry(peerEndpoint);
+
+        assertFalse(entryExistsInPeersV2(peerEndpoint));
+        assertFalse(entryExistsInLegacyPeers(peerEndpoint.getAddress()));
+
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertTrue("Missing peer should be added to peers_v2", entryExistsInPeersV2(peerEndpoint));
+        assertTrue("Missing peer should be added to legacy peers", entryExistsInLegacyPeers(peerEndpoint.getAddress()));
+    }
+
+    @Test
+    public void testStaleFieldInPeersV2IsRepaired()

Review Comment:
   Is it possible to add similar verification for every field in the table, preferably without too much duplication in the test code?



##########
test/unit/org/apache/cassandra/db/SystemPeersValidatorTest.java:
##########
@@ -0,0 +1,265 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.HashSet;
+import java.util.UUID;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.ServerTestUtils;
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.distributed.test.log.CMSTestBase;
+import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
+import org.apache.cassandra.harry.model.TokenPlacementModel;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.schema.SchemaConstants;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.tcm.AtomicLongBackedProcessor;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.ClusterMetadataService;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class SystemPeersValidatorTest
+{
+    private CMSTestBase.CMSSut sut;
+    private InetAddressAndPort peerEndpoint;
+
+    @BeforeClass
+    public static void beforeClass()
+    {
+        ServerTestUtils.prepareServerNoRegister();
+    }
+
+    @Before
+    public void before() throws Exception
+    {
+        ClusterMetadataService.unsetInstance();
+        sut = new CMSTestBase.CMSSut(AtomicLongBackedProcessor::new, false,
+                                     new TokenPlacementModel.SimpleReplicationFactor(3));
+        ClusterMetadataTestHelper.register(2);
+        ClusterMetadataTestHelper.join(2, 2);
+        peerEndpoint = ClusterMetadata.current().directory.endpoint(ClusterMetadataTestHelper.nodeId(2));
+        cleanupPeersTables();
+    }
+
+    @After
+    public void after()
+    {
+        if (sut != null)
+            sut.close();
+    }
+
+    @Test
+    public void testNoRepairWhenTablesAreConsistent()
+    {
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertEquals("peers_v2 should have 1 peer", 1, countEntries(PEERS_V2));
+        assertEquals("legacy peers should have 1 peer", 1, countEntries(LEGACY_PEERS));
+
+        // Second call should be a no-op
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertEquals("peers_v2 count should not change on second call", 1, countEntries(PEERS_V2));
+        assertEquals("legacy peers count should not change on second call", 1, countEntries(LEGACY_PEERS));
+    }
+
+    @Test
+    public void testStaleEntriesAreRemovedFromBothTables() throws UnknownHostException
+    {
+        InetAddressAndPort staleEndpoint = InetAddressAndPort.getByName("127.0.0.99");
+        insertStalePeerEntry(staleEndpoint, PEERS_V2);
+        insertStalePeerEntry(staleEndpoint, LEGACY_PEERS);
+
+        assertTrue(entryExistsInPeersV2(staleEndpoint));
+        assertTrue(entryExistsInLegacyPeers(staleEndpoint.getAddress()));
+
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertFalse("Stale entry should be removed from peers_v2", entryExistsInPeersV2(staleEndpoint));
+        assertFalse("Stale entry should be removed from legacy peers", entryExistsInLegacyPeers(staleEndpoint.getAddress()));
+    }
+
+    @Test
+    public void testStaleEntryOnlyInPeersV2IsRemoved() throws UnknownHostException
+    {
+        InetAddressAndPort staleEndpoint = InetAddressAndPort.getByName("127.0.0.99");
+        insertStalePeerEntry(staleEndpoint, PEERS_V2);
+
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertFalse("Stale entry should be removed from peers_v2", entryExistsInPeersV2(staleEndpoint));
+    }
+
+    @Test
+    public void testStaleEntryOnlyInLegacyPeersIsRemoved() throws UnknownHostException
+    {
+        InetAddressAndPort staleEndpoint = InetAddressAndPort.getByName("127.0.0.99");
+        insertStalePeerEntry(staleEndpoint, LEGACY_PEERS);
+
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertFalse("Stale entry should be removed from legacy peers", entryExistsInLegacyPeers(staleEndpoint.getAddress()));
+    }
+
+    @Test
+    public void testMissingPeerIsAddedToBothTables()
+    {
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+        removePeerEntry(peerEndpoint);
+
+        assertFalse(entryExistsInPeersV2(peerEndpoint));
+        assertFalse(entryExistsInLegacyPeers(peerEndpoint.getAddress()));
+
+        SystemPeersValidator.validateAndRepair(ClusterMetadata.current());
+
+        assertTrue("Missing peer should be added to peers_v2", entryExistsInPeersV2(peerEndpoint));
+        assertTrue("Missing peer should be added to legacy peers", entryExistsInLegacyPeers(peerEndpoint.getAddress()));
+    }
+
+    @Test
+    public void testStaleFieldInPeersV2IsRepaired()

Review Comment:
   Relatedly, what if the mismatch is due to a null column in the system table?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] Make sure system.peers and peers_v2 match cluster metadata [cassandra]

["minal-kyada (via GitHub)" <gi...@apache.org>]

minal-kyada commented on code in PR #4630:
URL: https://github.com/apache/cassandra/pull/4630#discussion_r2884789658


##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+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.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata on startup.
+ * This is critical for backward compatibility as older clients and tools read from these
+ * legacy tables while TCM uses ClusterMetadata as the source of truth.
+ *
+ * The validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ */
+public class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        if (metadata != null)
+        {
+            Set<NodeId> expectedNodes = getExpectedPeerNodes(metadata);
+            Set<InetAddress> legacyPeersEntries = getLegacyPeersEntries();
+
+            Map<InetAddressAndPort, NodeId> expectedEndpoints = new HashMap<>();
+            Map<InetAddress, NodeId> expectedAddresses = new HashMap<>();
+            for (NodeId nodeId : expectedNodes)
+            {
+                InetAddressAndPort endpoint = metadata.directory.endpoint(nodeId);
+                expectedEndpoints.put(endpoint, nodeId);
+                expectedAddresses.put(endpoint.getAddress(), nodeId);
+            }
+
+            Set<InetAddressAndPort> peersV2Entries = getPeersV2Entries();
+            for (Map.Entry<InetAddressAndPort, NodeId> entry : expectedEndpoints.entrySet())
+            {
+                InetAddressAndPort endpoint = entry.getKey();
+                NodeId nodeId = entry.getValue();
+
+                boolean inPeersV2 = peersV2Entries.contains(endpoint);
+                boolean inLegacyPeers = legacyPeersEntries.contains(endpoint.getAddress());
+
+                if (!inPeersV2 || !inLegacyPeers)
+                {
+                    logger.info("Repairing missing peer entry for {} (nodeId={}, inPeersV2={}, inLegacyPeers={})",
+                                endpoint, nodeId, inPeersV2, inLegacyPeers);
+                    PeersTable.updateLegacyPeerTable(nodeId, metadata, metadata);
+                }
+            }
+
+            String deleteV2Query = String.format("DELETE FROM %s.%s WHERE peer = ? AND peer_port = ?",
+                                                 SchemaConstants.SYSTEM_KEYSPACE_NAME, PEERS_V2);
+            for (InetAddressAndPort endpoint : peersV2Entries)
+            {
+                if (!expectedEndpoints.containsKey(endpoint))
+                {
+                    logger.info("Removing stale entry from {}: endpoint {} not found in ClusterMetadata",
+                                PEERS_V2, endpoint);
+                    executeInternal(deleteV2Query, endpoint.getAddress(), endpoint.getPort());
+                }
+            }
+
+            String deleteQuery = String.format("DELETE FROM %s.%s WHERE peer = ?",
+                                                 SchemaConstants.SYSTEM_KEYSPACE_NAME, LEGACY_PEERS);
+            for (InetAddress address : legacyPeersEntries)
+            {
+                if (!expectedAddresses.containsKey(address))
+                {
+                    logger.info("Removing stale entry from {}: address {} not found in ClusterMetadata",
+                                LEGACY_PEERS, address);
+                    executeInternal(deleteQuery, address);
+                }
+            }
+        }
+    }
+
+    private static Set<NodeId> getExpectedPeerNodes(ClusterMetadata metadata)
+    {
+        Set<NodeId> expectedNodes = new HashSet<>();
+        InetAddressAndPort localEndpoint = FBUtilities.getBroadcastAddressAndPort();
+
+        for (InetAddressAndPort endpoint : metadata.directory.allJoinedEndpoints())
+        {
+            if (endpoint.equals(localEndpoint)) continue;
+            expectedNodes.add(metadata.directory.peerId(endpoint));
+        }
+        return expectedNodes;
+    }
+
+    private static Set<InetAddressAndPort> getPeersV2Entries()
+    {
+        Set<InetAddressAndPort> entries = new HashSet<>();
+        String query = String.format("SELECT peer, peer_port FROM %s.%s", SchemaConstants.SYSTEM_KEYSPACE_NAME, PEERS_V2);

Review Comment:
   Yes, we should! Thanks Alex.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] Make sure system.peers and peers_v2 match cluster metadata [cassandra]

["minal-kyada (via GitHub)" <gi...@apache.org>]

minal-kyada commented on code in PR #4630:
URL: https://github.com/apache/cassandra/pull/4630#discussion_r2897648505


##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.UTF8Type;
+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.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata on startup.
+ * This is critical for backward compatibility as older clients and tools read from these
+ * legacy tables while TCM uses ClusterMetadata as the source of truth.
+ *
+ * The validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ */
+public class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        Map<InetAddressAndPort, UntypedResultSet.Row> peersV2Rows = getPeersV2Rows();
+        Map<InetAddress, UntypedResultSet.Row> legacyPeersRows = getLegacyPeersRows();
+
+        Map<InetAddressAndPort, NodeId> expectedEndpoints = new HashMap<>();
+        Map<InetAddress, NodeId> expectedAddresses = new HashMap<>();
+        for (NodeId nodeId : getExpectedPeerNodes(metadata))

Review Comment:
   I had this because I wanted to eliminate localEndpoints, but there's a better way to do it - Inlined the logic directly into validateAndRepair, iterating allJoinedEndpoints() once and looking up nodeId from endpoint directly and skipping the localEndpoint.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] Make sure system.peers and peers_v2 match cluster metadata [cassandra]

["minal-kyada (via GitHub)" <gi...@apache.org>]

minal-kyada commented on PR #4630:
URL: https://github.com/apache/cassandra/pull/4630#issuecomment-4644248266

   https://issues.apache.org/jira/browse/CASSANDRA-21187 
   Closing the PR, as the fix is applied and the Jira ticket is marked resolved!


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] Make sure system.peers and peers_v2 match cluster metadata [cassandra]

["minal-kyada (via GitHub)" <gi...@apache.org>]

minal-kyada commented on code in PR #4630:
URL: https://github.com/apache/cassandra/pull/4630#discussion_r2898091524


##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.UTF8Type;
+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.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata on startup.
+ * This is critical for backward compatibility as older clients and tools read from these
+ * legacy tables while TCM uses ClusterMetadata as the source of truth.
+ *
+ * The validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ */
+public class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        Map<InetAddressAndPort, UntypedResultSet.Row> peersV2Rows = getPeersV2Rows();
+        Map<InetAddress, UntypedResultSet.Row> legacyPeersRows = getLegacyPeersRows();
+
+        Map<InetAddressAndPort, NodeId> expectedEndpoints = new HashMap<>();
+        Map<InetAddress, NodeId> expectedAddresses = new HashMap<>();
+        for (NodeId nodeId : getExpectedPeerNodes(metadata))
+        {
+            InetAddressAndPort endpoint = metadata.directory.endpoint(nodeId);
+            expectedEndpoints.put(endpoint, nodeId);
+            expectedAddresses.put(endpoint.getAddress(), nodeId);
+        }
+
+        String deleteV2Query = String.format("DELETE FROM %s.%s WHERE peer = ? AND peer_port = ?",
+                                               SchemaConstants.SYSTEM_KEYSPACE_NAME, PEERS_V2);
+        for (InetAddressAndPort endpoint : peersV2Rows.keySet())
+        {
+            if (!expectedEndpoints.containsKey(endpoint))
+            {
+                logger.info("Removing stale peer {} from {}", endpoint, PEERS_V2);
+                executeInternal(deleteV2Query, endpoint.getAddress(), endpoint.getPort());
+            }
+        }
+
+        String deleteLegacyQuery = String.format("DELETE FROM %s.%s WHERE peer = ?",
+                                                   SchemaConstants.SYSTEM_KEYSPACE_NAME,
+                                                   LEGACY_PEERS);
+        for (InetAddress address : legacyPeersRows.keySet())
+        {
+            if (!expectedAddresses.containsKey(address))
+            {
+                logger.info("Removing stale peer {} from {}", address, LEGACY_PEERS);
+                executeInternal(deleteLegacyQuery, address);
+            }
+        }
+
+        for (Map.Entry<InetAddressAndPort, NodeId> entry : expectedEndpoints.entrySet())
+        {
+            NodeId nodeId = entry.getValue();
+            InetAddressAndPort endpoint = entry.getKey();
+            UntypedResultSet.Row v2Row = peersV2Rows.get(endpoint);
+            UntypedResultSet.Row legacyRow = legacyPeersRows.get(endpoint.getAddress());
+
+            List<String> v2Discrepancies = collectV2Discrepancies(v2Row, nodeId, metadata);

Review Comment:
   > null safety
    
   Thanks for sharing this, I should definitely add it, as except for `getSet()`, none of the getters have null guards. I'll add `has()` checks before each getter and treat a missing column as a discrepancy.    
   
   > consolidating the two methods
   
   I did consider it, but reverted that change because I found it more convenient to see the complete set of validations for a given table at a glance without jumping between methods. A shared helper would still need table-specific checks alongside it, so I'd end up fragmenting the logic across multiple methods for what is a fairly simple use case. That said, happy to revisit if you feel strongly about it.
   
   > per-field discrepancy logging
   
   It tells an operator exactly what's wrong — e.g. `Updating peer 10.0.0.1 in peers_v2 for stale fields [data_center, rack]`. With Row::toString(), we will get pipe-delimited positional values without column names, and the operator would have to figure out the diff manually to identify what's stale. For something meant to be a diagnostic/repair tool, don't you think having the specific out-of-sync fields is more helpful? I also checked the rest of the codebase and we always extract fields and log them by name. Please let me know if it is useful to know this information from operator POV, else we can definitely add this change ?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] Make sure system.peers and peers_v2 match cluster metadata [cassandra]

["minal-kyada (via GitHub)" <gi...@apache.org>]

minal-kyada closed pull request #4630: Make sure system.peers and peers_v2 match cluster metadata                      
URL: https://github.com/apache/cassandra/pull/4630


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] Make sure system.peers and peers_v2 match cluster metadata [cassandra]

["beobal (via GitHub)" <gi...@apache.org>]

beobal commented on code in PR #4630:
URL: https://github.com/apache/cassandra/pull/4630#discussion_r2904014619


##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.UTF8Type;
+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.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata on startup.
+ * This is critical for backward compatibility as older clients and tools read from these
+ * legacy tables while TCM uses ClusterMetadata as the source of truth.
+ *
+ * The validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ */
+public class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        Map<InetAddressAndPort, UntypedResultSet.Row> peersV2Rows = getPeersV2Rows();
+        Map<InetAddress, UntypedResultSet.Row> legacyPeersRows = getLegacyPeersRows();
+
+        Map<InetAddressAndPort, NodeId> expectedEndpoints = new HashMap<>();
+        Map<InetAddress, NodeId> expectedAddresses = new HashMap<>();
+        for (NodeId nodeId : getExpectedPeerNodes(metadata))
+        {
+            InetAddressAndPort endpoint = metadata.directory.endpoint(nodeId);
+            expectedEndpoints.put(endpoint, nodeId);
+            expectedAddresses.put(endpoint.getAddress(), nodeId);
+        }
+
+        String deleteV2Query = String.format("DELETE FROM %s.%s WHERE peer = ? AND peer_port = ?",
+                                               SchemaConstants.SYSTEM_KEYSPACE_NAME, PEERS_V2);
+        for (InetAddressAndPort endpoint : peersV2Rows.keySet())
+        {
+            if (!expectedEndpoints.containsKey(endpoint))
+            {
+                logger.info("Removing stale peer {} from {}", endpoint, PEERS_V2);
+                executeInternal(deleteV2Query, endpoint.getAddress(), endpoint.getPort());
+            }
+        }
+
+        String deleteLegacyQuery = String.format("DELETE FROM %s.%s WHERE peer = ?",
+                                                   SchemaConstants.SYSTEM_KEYSPACE_NAME,
+                                                   LEGACY_PEERS);
+        for (InetAddress address : legacyPeersRows.keySet())
+        {
+            if (!expectedAddresses.containsKey(address))
+            {
+                logger.info("Removing stale peer {} from {}", address, LEGACY_PEERS);
+                executeInternal(deleteLegacyQuery, address);
+            }
+        }
+
+        for (Map.Entry<InetAddressAndPort, NodeId> entry : expectedEndpoints.entrySet())
+        {
+            NodeId nodeId = entry.getValue();
+            InetAddressAndPort endpoint = entry.getKey();
+            UntypedResultSet.Row v2Row = peersV2Rows.get(endpoint);
+            UntypedResultSet.Row legacyRow = legacyPeersRows.get(endpoint.getAddress());
+
+            List<String> v2Discrepancies = collectV2Discrepancies(v2Row, nodeId, metadata);

Review Comment:
   > It tells an operator exactly what's wrong — e.g. Updating peer 10.0.0.1 in peers_v2 for stale fields [data_center, rack]
   
   yes, IMO this is unnecessary and given the frequency with which I would anticipate this actually happening, I don't think it's worth the code bloat.
   
   > With Row::toString(), we will get pipe-delimited positional values without column names, and the operator would have to figure out the diff manually to identify what's stale.
   
   I think this is fine, in the unlikely situation that a mismatch is detected having the full row logged would be good enough, figuring out the diff manually would pretty straightforward. 
   
   > I did consider it, but reverted that change because I found it more convenient to see the complete set of validations for a given table at a glance without jumping between methods. 
   
   What I had in mind was something like:
   ```
       private static boolean peersV2RowIsEquivalent(UntypedResultSet.Row row,
                                                     NodeId nodeId,
                                                     ClusterMetadata metadata)
       {
           NodeAddresses addresses = metadata.directory.getNodeAddresses(nodeId);
           return commonColumnsAreEquivalent(row, nodeId, addresses, metadata) &&
                  row.has("preferred_port") && Objects.equals(row.getInetAddress("preferred_port"), addresses.broadcastAddress.getPort()) &&
                  row.has("rpc_port") && Objects.equals(row.getInetAddress("rpc_port"), addresses.nativeAddress.getPort());
       }
   
       private static boolean peersRowIsEquivalent(UntypedResultSet.Row row,
                                                   NodeId nodeId,
                                                   ClusterMetadata metadata)
       {
           NodeAddresses addresses = metadata.directory.getNodeAddresses(nodeId);
           return commonColumnsAreEquivalent(row, nodeId, addresses, metadata);
       }
   
       private static boolean commonColumnsAreEquivalent(UntypedResultSet.Row row,
                                                         NodeId nodeId,
                                                         NodeAddresses addresses,
                                                         ClusterMetadata metadata)
       {
           if (row == null)
               return false;
           Location location = metadata.directory.location(nodeId);
           // This column is differently named in the peers and peers_v2 tables
           String nativeAddressColumn = row.has("native_address") ? "native_address" : "rpc_address";
   
           // Check existence first because row.getXXX can NPE if the column is not present
           return row.has("preferred_ip")      && Objects.equals(row.getInetAddress("preferred_ip"), addresses.broadcastAddress.getAddress()) &&
                  row.has(nativeAddressColumn) && Objects.equals(row.getInetAddress(nativeAddressColumn), addresses.nativeAddress.getAddress()) &&
                  row.has("data_center")       && Objects.equals(row.getString("data_center"), location.datacenter) &&
                  row.has("rack")              && Objects.equals(row.getString("rack"), location.rack) &&
                  row.has("host_id")           && Objects.equals(row.getUUID("host_id"), nodeId.toUUID()) &&
                  row.has("release_version")   && Objects.equals(row.getString("release_version"), metadata.directory.version(nodeId)) &&
                  row.has("schema_version")    && Objects.equals(row.getUUID("schema_version"), metadata.schema.getVersion()) &&
                  row.has("tokens")            && Objects.equals(row.getSet("tokens", UTF8Type.instance), tokensAsSet(metadata.tokenMap.tokens(nodeId)));
       }
   ```
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] Make sure system.peers and peers_v2 match cluster metadata [cassandra]

["minal-kyada (via GitHub)" <gi...@apache.org>]

minal-kyada commented on code in PR #4630:
URL: https://github.com/apache/cassandra/pull/4630#discussion_r2906055529


##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.UTF8Type;
+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.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata on startup.
+ * This is critical for backward compatibility as older clients and tools read from these
+ * legacy tables while TCM uses ClusterMetadata as the source of truth.
+ *
+ * The validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ */
+public class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        Map<InetAddressAndPort, UntypedResultSet.Row> peersV2Rows = getPeersV2Rows();
+        Map<InetAddress, UntypedResultSet.Row> legacyPeersRows = getLegacyPeersRows();
+
+        Map<InetAddressAndPort, NodeId> expectedEndpoints = new HashMap<>();
+        Map<InetAddress, NodeId> expectedAddresses = new HashMap<>();
+        for (NodeId nodeId : getExpectedPeerNodes(metadata))
+        {
+            InetAddressAndPort endpoint = metadata.directory.endpoint(nodeId);
+            expectedEndpoints.put(endpoint, nodeId);
+            expectedAddresses.put(endpoint.getAddress(), nodeId);
+        }
+
+        String deleteV2Query = String.format("DELETE FROM %s.%s WHERE peer = ? AND peer_port = ?",
+                                               SchemaConstants.SYSTEM_KEYSPACE_NAME, PEERS_V2);
+        for (InetAddressAndPort endpoint : peersV2Rows.keySet())
+        {
+            if (!expectedEndpoints.containsKey(endpoint))
+            {
+                logger.info("Removing stale peer {} from {}", endpoint, PEERS_V2);
+                executeInternal(deleteV2Query, endpoint.getAddress(), endpoint.getPort());
+            }
+        }
+
+        String deleteLegacyQuery = String.format("DELETE FROM %s.%s WHERE peer = ?",
+                                                   SchemaConstants.SYSTEM_KEYSPACE_NAME,
+                                                   LEGACY_PEERS);
+        for (InetAddress address : legacyPeersRows.keySet())
+        {
+            if (!expectedAddresses.containsKey(address))
+            {
+                logger.info("Removing stale peer {} from {}", address, LEGACY_PEERS);
+                executeInternal(deleteLegacyQuery, address);
+            }
+        }
+
+        for (Map.Entry<InetAddressAndPort, NodeId> entry : expectedEndpoints.entrySet())
+        {
+            NodeId nodeId = entry.getValue();
+            InetAddressAndPort endpoint = entry.getKey();
+            UntypedResultSet.Row v2Row = peersV2Rows.get(endpoint);
+            UntypedResultSet.Row legacyRow = legacyPeersRows.get(endpoint.getAddress());
+
+            List<String> v2Discrepancies = collectV2Discrepancies(v2Row, nodeId, metadata);

Review Comment:
   Sure I can make these changes if figuring out the diff manually is no big deal. And about having a helper method for common field validations, I did the same thing. I can push that change too. Thanks!



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] Make sure system.peers and peers_v2 match cluster metadata [cassandra]

["ifesdjeen (via GitHub)" <gi...@apache.org>]

ifesdjeen commented on code in PR #4630:
URL: https://github.com/apache/cassandra/pull/4630#discussion_r2851973543


##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+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.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata on startup.
+ * This is critical for backward compatibility as older clients and tools read from these
+ * legacy tables while TCM uses ClusterMetadata as the source of truth.
+ *
+ * The validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ */
+public class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        if (metadata != null)
+        {
+            Set<NodeId> expectedNodes = getExpectedPeerNodes(metadata);
+            Set<InetAddress> legacyPeersEntries = getLegacyPeersEntries();
+
+            Map<InetAddressAndPort, NodeId> expectedEndpoints = new HashMap<>();
+            Map<InetAddress, NodeId> expectedAddresses = new HashMap<>();
+            for (NodeId nodeId : expectedNodes)
+            {
+                InetAddressAndPort endpoint = metadata.directory.endpoint(nodeId);
+                expectedEndpoints.put(endpoint, nodeId);
+                expectedAddresses.put(endpoint.getAddress(), nodeId);
+            }
+
+            Set<InetAddressAndPort> peersV2Entries = getPeersV2Entries();
+            for (Map.Entry<InetAddressAndPort, NodeId> entry : expectedEndpoints.entrySet())
+            {
+                InetAddressAndPort endpoint = entry.getKey();
+                NodeId nodeId = entry.getValue();
+
+                boolean inPeersV2 = peersV2Entries.contains(endpoint);
+                boolean inLegacyPeers = legacyPeersEntries.contains(endpoint.getAddress());
+
+                if (!inPeersV2 || !inLegacyPeers)
+                {
+                    logger.info("Repairing missing peer entry for {} (nodeId={}, inPeersV2={}, inLegacyPeers={})",
+                                endpoint, nodeId, inPeersV2, inLegacyPeers);
+                    PeersTable.updateLegacyPeerTable(nodeId, metadata, metadata);
+                }
+            }
+
+            String deleteV2Query = String.format("DELETE FROM %s.%s WHERE peer = ? AND peer_port = ?",
+                                                 SchemaConstants.SYSTEM_KEYSPACE_NAME, PEERS_V2);
+            for (InetAddressAndPort endpoint : peersV2Entries)
+            {
+                if (!expectedEndpoints.containsKey(endpoint))
+                {
+                    logger.info("Removing stale entry from {}: endpoint {} not found in ClusterMetadata",
+                                PEERS_V2, endpoint);
+                    executeInternal(deleteV2Query, endpoint.getAddress(), endpoint.getPort());
+                }
+            }
+
+            String deleteQuery = String.format("DELETE FROM %s.%s WHERE peer = ?",
+                                                 SchemaConstants.SYSTEM_KEYSPACE_NAME, LEGACY_PEERS);
+            for (InetAddress address : legacyPeersEntries)
+            {
+                if (!expectedAddresses.containsKey(address))
+                {
+                    logger.info("Removing stale entry from {}: address {} not found in ClusterMetadata",
+                                LEGACY_PEERS, address);
+                    executeInternal(deleteQuery, address);
+                }
+            }
+        }
+    }
+
+    private static Set<NodeId> getExpectedPeerNodes(ClusterMetadata metadata)
+    {
+        Set<NodeId> expectedNodes = new HashSet<>();
+        InetAddressAndPort localEndpoint = FBUtilities.getBroadcastAddressAndPort();
+
+        for (InetAddressAndPort endpoint : metadata.directory.allJoinedEndpoints())
+        {
+            if (endpoint.equals(localEndpoint)) continue;
+            expectedNodes.add(metadata.directory.peerId(endpoint));
+        }
+        return expectedNodes;
+    }
+
+    private static Set<InetAddressAndPort> getPeersV2Entries()
+    {
+        Set<InetAddressAndPort> entries = new HashSet<>();
+        String query = String.format("SELECT peer, peer_port FROM %s.%s", SchemaConstants.SYSTEM_KEYSPACE_NAME, PEERS_V2);

Review Comment:
   Don't we want to validate _all_ fields of the table? In `updateLegacyPeerTable`, we're updating more than just these two fields:
   
   ```
               QueryProcessor.executeInternal(String.format(peers_v2_query, SYSTEM_KEYSPACE_NAME, PEERS_V2),
                                              addresses.broadcastAddress.getAddress(), addresses.broadcastAddress.getPort(),
                                              addresses.broadcastAddress.getAddress(), addresses.broadcastAddress.getPort(),
                                              addresses.nativeAddress.getAddress(), addresses.nativeAddress.getPort(),
                                              location.datacenter, location.rack,
                                              nodeId.toUUID(),
                                              next.directory.version(nodeId).cassandraVersion.toString(),
                                              next.schema.getVersion(),
                                              tokens);
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] Make sure system.peers and peers_v2 match cluster metadata [cassandra]

["ifesdjeen (via GitHub)" <gi...@apache.org>]

ifesdjeen commented on code in PR #4630:
URL: https://github.com/apache/cassandra/pull/4630#discussion_r2851956830


##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+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.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata on startup.
+ * This is critical for backward compatibility as older clients and tools read from these
+ * legacy tables while TCM uses ClusterMetadata as the source of truth.
+ *
+ * The validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ */
+public class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        if (metadata != null)

Review Comment:
   we should probably throw if it's null; 
   
   also, a small nit: I wouldn't nest entire method within `if`'s brackets, i would early-exit on unmatched condition, and unindent the rest of the method.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] Make sure system.peers and peers_v2 match cluster metadata [cassandra]

["minal-kyada (via GitHub)" <gi...@apache.org>]

minal-kyada commented on code in PR #4630:
URL: https://github.com/apache/cassandra/pull/4630#discussion_r2897648505


##########
src/java/org/apache/cassandra/db/SystemPeersValidator.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.db;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.UTF8Type;
+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.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_PEERS;
+import static org.apache.cassandra.db.SystemKeyspace.PEERS_V2;
+
+/**
+ * Validator to ensure system.peers and system.peers_v2 tables match ClusterMetadata on startup.
+ * This is critical for backward compatibility as older clients and tools read from these
+ * legacy tables while TCM uses ClusterMetadata as the source of truth.
+ *
+ * The validator detects inconsistencies and automatically repairs them by synchronizing
+ * the peers tables with the current ClusterMetadata.
+ */
+public class SystemPeersValidator
+{
+    private static final Logger logger = LoggerFactory.getLogger(SystemPeersValidator.class);
+
+    public static void validateAndRepair(ClusterMetadata metadata)
+    {
+        Map<InetAddressAndPort, UntypedResultSet.Row> peersV2Rows = getPeersV2Rows();
+        Map<InetAddress, UntypedResultSet.Row> legacyPeersRows = getLegacyPeersRows();
+
+        Map<InetAddressAndPort, NodeId> expectedEndpoints = new HashMap<>();
+        Map<InetAddress, NodeId> expectedAddresses = new HashMap<>();
+        for (NodeId nodeId : getExpectedPeerNodes(metadata))

Review Comment:
   I had this because I wanted to eliminate localEndpoints, but there's a better way to do it - Inlined the logic directly into validateAndRepair, iterating
   allJoinedEndpoints() once and looking up nodeId from endpoint directly and skipping the localEndpoint.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org