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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
