From c484fc511accda38bf3fcf6906cb58b39ab4503f Mon Sep 17 00:00:00 2001 From: Minal Kyada Date: Mon, 23 Feb 2026 16:02:57 -0800 Subject: [PATCH] Add tooling to repair system peers tables if inconsistent with cluster metadata Patch by Minal Kyada; reviewed by Alex Petrov and Sam Tunnicliffe for CASSANDRA-21187 --- CHANGES.txt | 1 + .../config/CassandraRelevantProperties.java | 1 + .../cassandra/db/SystemPeersValidator.java | 185 ++++++++++ .../cassandra/service/StorageService.java | 10 + .../service/StorageServiceMBean.java | 9 + .../db/SystemPeersValidatorTest.java | 322 ++++++++++++++++++ 6 files changed, 528 insertions(+) create mode 100644 src/java/org/apache/cassandra/db/SystemPeersValidator.java create mode 100644 test/unit/org/apache/cassandra/db/SystemPeersValidatorTest.java diff --git a/CHANGES.txt b/CHANGES.txt index 83b3bb7d67..d64329acbc 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 6.0-alpha2 + * Add tooling to repair system peers and peers_v2 if inconsistent with cluster metadata (CASSANDRA-21187) * Fix a removed TTLed row re-appearance in a materialized view after a cursor compaction (CASSANDRA-21152) * Rework ZSTD dictionary compression logic to create a trainer per training (CASSANDRA-21209) Merged from 5.0: diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index 9df5e2f201..8a60d44797 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -587,6 +587,7 @@ public enum CassandraRelevantProperties SUN_STDERR_ENCODING("sun.stderr.encoding"), SUN_STDOUT_ENCODING("sun.stdout.encoding"), SUPERUSER_SETUP_DELAY_MS("cassandra.superuser_setup_delay_ms", "10000"), + SYNC_SYSTEM_PEERS_TABLES_AT_STARTUP("cassandra.sync_system_peers_tables_at_startup", "true"), SYSTEM_AUTH_DEFAULT_RF("cassandra.system_auth.default_rf", "1"), SYSTEM_DISTRIBUTED_DEFAULT_RF("cassandra.system_distributed.default_rf", "3"), SYSTEM_TRACES_DEFAULT_RF("cassandra.system_traces.default_rf", "2"), diff --git a/src/java/org/apache/cassandra/db/SystemPeersValidator.java b/src/java/org/apache/cassandra/db/SystemPeersValidator.java new file mode 100644 index 0000000000..2f490a4bb4 --- /dev/null +++ b/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 peersV2Rows = getPeersV2Rows(); + Map peersRows = getPeersRows(); + + Map knownEndpoints = new HashMap<>(); + Set 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 system.{}", 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 system.{}", address, LEGACY_PEERS); + executeInternal(DELETE_PEERS_QUERY, address); + } + } + + for (Map.Entry 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 system.{}", 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 system.{}: {}", 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"; + + // 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).cassandraVersion.toString()) + && row.has("schema_version") && Objects.equals(row.getUUID("schema_version"), metadata.schema.getVersion()) + && row.has("tokens") && Objects.equals(row.getSet("tokens", UTF8Type.instance), SystemKeyspace.tokensAsSet(metadata.tokenMap.tokens(nodeId))); + } + + private static Map getPeersV2Rows() + { + Map rows = new HashMap<>(); + for (UntypedResultSet.Row row : executeInternal(SELECT_PEERS_V2_QUERY)) + rows.put(InetAddressAndPort.getByAddressOverrideDefaults(row.getInetAddress("peer"), + row.getInt("peer_port")), row); + return rows; + } + + private static Map getPeersRows() + { + Map rows = new HashMap<>(); + for (UntypedResultSet.Row row : executeInternal(SELECT_PEERS_QUERY)) + rows.put(row.getInetAddress("peer"), row); + return rows; + } +} diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 424df75b97..b7460b1be5 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -103,6 +103,7 @@ import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.db.SizeEstimatesRecorder; import org.apache.cassandra.db.SystemKeyspace; +import org.apache.cassandra.db.SystemPeersValidator; import org.apache.cassandra.db.commitlog.CommitLog; import org.apache.cassandra.db.compaction.CompactionManager; import org.apache.cassandra.db.compaction.OperationType; @@ -859,6 +860,9 @@ public class StorageService extends NotificationBroadcasterSupport implements IE RegistrationStatus.instance.onRegistration(); Startup.maybeExecuteStartupTransformation(self); + if (CassandraRelevantProperties.SYNC_SYSTEM_PEERS_TABLES_AT_STARTUP.getBoolean()) + SystemPeersValidator.validateAndRepair(ClusterMetadata.current()); + try { if (joinRing) @@ -5802,6 +5806,12 @@ public class StorageService extends NotificationBroadcasterSupport implements IE return Keyspace.open(keyspace).getColumnFamilyStores().stream().map(cfs -> cfs.name).collect(Collectors.toList()); } + @Override + public void validateAndRepairPeersMetadata() + { + SystemPeersValidator.validateAndRepair(ClusterMetadata.current()); + } + @Override public List mutateSSTableRepairedState(boolean repaired, boolean preview, String keyspace, List tableNames) { diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index 058bc8955f..b9f47b3b2d 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -1410,6 +1410,15 @@ public interface StorageServiceMBean extends NotificationEmitter /** Gets the names of all tables for the given keyspace */ public List getTablesForKeyspace(String keyspace); + /** + * Validates that system.peers and system.peers_v2 are consistent with ClusterMetadata, + * inserting missing peer entries and removing stale ones. This runs automatically on + * startup but can be triggered manually if a discrepancy is suspected. + *

+ * Note: mutates data in system.peers and system.peers_v2. + */ + public void validateAndRepairPeersMetadata(); + /** Mutates the repaired state of all SSTables for the given SSTables */ public List mutateSSTableRepairedState(boolean repaired, boolean preview, String keyspace, List tables); diff --git a/test/unit/org/apache/cassandra/db/SystemPeersValidatorTest.java b/test/unit/org/apache/cassandra/db/SystemPeersValidatorTest.java new file mode 100644 index 0000000000..796de62642 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/SystemPeersValidatorTest.java @@ -0,0 +1,322 @@ +/* + * 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.db.marshal.UTF8Type; +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.service.StorageService; +import org.apache.cassandra.tcm.AtomicLongBackedProcessor; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.ClusterMetadataService; +import org.apache.cassandra.tcm.membership.Location; +import org.apache.cassandra.tcm.membership.NodeAddresses; +import org.apache.cassandra.tcm.membership.NodeId; + +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.apache.cassandra.schema.SchemaConstants.SYSTEM_KEYSPACE_NAME; +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("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("peers count should not change on second call", 1, countEntries(LEGACY_PEERS)); + } + + @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 testStaleEntryOnlyInPeersIsRemoved() 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 peers", entryExistsInPeers(staleEndpoint.getAddress())); + } + + @Test + public void testMissingPeerIsAddedToBothTables() + { + SystemPeersValidator.validateAndRepair(ClusterMetadata.current()); + removePeerEntry(peerEndpoint); + + assertFalse(entryExistsInPeersV2(peerEndpoint)); + assertFalse(entryExistsInPeers(peerEndpoint.getAddress())); + + SystemPeersValidator.validateAndRepair(ClusterMetadata.current()); + + assertPeersV2RowMatchesMetadata(peerEndpoint); + assertPeersRowMatchesMetadata(peerEndpoint); + } + + @Test + public void testStaleFieldInPeersV2IsRepaired() + { + SystemPeersValidator.validateAndRepair(ClusterMetadata.current()); + + executeInternal(String.format("UPDATE %s.%s SET data_center = 'stale-dc' WHERE peer = ? AND peer_port = ?", + SYSTEM_KEYSPACE_NAME, PEERS_V2), + peerEndpoint.getAddress(), peerEndpoint.getPort()); + + SystemPeersValidator.validateAndRepair(ClusterMetadata.current()); + + assertPeersV2RowMatchesMetadata(peerEndpoint); + } + + @Test + public void testStaleFieldInPeersIsRepaired() + { + SystemPeersValidator.validateAndRepair(ClusterMetadata.current()); + + executeInternal(String.format("UPDATE %s.%s SET data_center = 'stale-dc' WHERE peer = ?", + SYSTEM_KEYSPACE_NAME, LEGACY_PEERS), + peerEndpoint.getAddress()); + + SystemPeersValidator.validateAndRepair(ClusterMetadata.current()); + + assertPeersRowMatchesMetadata(peerEndpoint); + } + + @Test + public void testNullColumnInPeersV2IsRepaired() + { + String[] columns = { "data_center", "host_id", "preferred_ip", "preferred_port", + "rack", "release_version", "native_address", "native_port", + "schema_version", "tokens" }; + + SystemPeersValidator.validateAndRepair(ClusterMetadata.current()); + + for (String column : columns) + { + executeInternal(String.format("DELETE %s FROM %s.%s WHERE peer = ? AND peer_port = ?", + column, SYSTEM_KEYSPACE_NAME, PEERS_V2), + peerEndpoint.getAddress(), peerEndpoint.getPort()); + + SystemPeersValidator.validateAndRepair(ClusterMetadata.current()); + + assertPeersV2RowMatchesMetadata(peerEndpoint); + } + } + + @Test + public void testNullColumnInPeersIsRepaired() + { + String[] columns = { "data_center", "host_id", "preferred_ip", + "rack", "release_version", "rpc_address", + "schema_version", "tokens" }; + + SystemPeersValidator.validateAndRepair(ClusterMetadata.current()); + + for (String column : columns) + { + executeInternal(String.format("DELETE %s FROM %s.%s WHERE peer = ?", + column, SYSTEM_KEYSPACE_NAME, LEGACY_PEERS), + peerEndpoint.getAddress()); + + SystemPeersValidator.validateAndRepair(ClusterMetadata.current()); + + assertPeersRowMatchesMetadata(peerEndpoint); + } + } + + @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 peers", entryExistsInPeers(peerEndpoint.getAddress())); + } + + private void cleanupPeersTables() + { + executeInternal(String.format("TRUNCATE %s.%s", SYSTEM_KEYSPACE_NAME, PEERS_V2)); + executeInternal(String.format("TRUNCATE %s.%s", SYSTEM_KEYSPACE_NAME, LEGACY_PEERS)); + } + + private int countEntries(String table) + { + UntypedResultSet result = executeInternal(String.format("SELECT COUNT(*) FROM %s.%s", + 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 = ?", + SYSTEM_KEYSPACE_NAME, PEERS_V2), + endpoint.getAddress(), endpoint.getPort()); + return !result.isEmpty(); + } + + private boolean entryExistsInPeers(InetAddress address) + { + UntypedResultSet result = executeInternal( + String.format("SELECT peer FROM %s.%s WHERE peer = ?", + SYSTEM_KEYSPACE_NAME, LEGACY_PEERS), + address); + return !result.isEmpty(); + } + + private void assertPeersV2RowMatchesMetadata(InetAddressAndPort endpoint) + { + ClusterMetadata metadata = ClusterMetadata.current(); + NodeId nodeId = ClusterMetadataTestHelper.nodeId(2); + NodeAddresses addresses = metadata.directory.getNodeAddresses(nodeId); + Location location = metadata.directory.location(nodeId); + + UntypedResultSet result = executeInternal( + String.format("SELECT * FROM %s.%s WHERE peer = ? AND peer_port = ?", + SYSTEM_KEYSPACE_NAME, PEERS_V2), + endpoint.getAddress(), endpoint.getPort()); + UntypedResultSet.Row row = result.one(); + + assertEquals(addresses.broadcastAddress.getAddress(), row.getInetAddress("preferred_ip")); + assertEquals(addresses.broadcastAddress.getPort(), row.getInt("preferred_port")); + assertEquals(addresses.nativeAddress.getAddress(), row.getInetAddress("native_address")); + assertEquals(addresses.nativeAddress.getPort(), row.getInt("native_port")); + assertEquals(location.datacenter, row.getString("data_center")); + assertEquals(location.rack, row.getString("rack")); + assertEquals(nodeId.toUUID(), row.getUUID("host_id")); + assertEquals(metadata.directory.version(nodeId).cassandraVersion.toString(), row.getString("release_version")); + assertEquals(metadata.schema.getVersion(), row.getUUID("schema_version")); + assertEquals(SystemKeyspace.tokensAsSet(metadata.tokenMap.tokens(nodeId)), row.getSet("tokens", UTF8Type.instance)); + } + + private void assertPeersRowMatchesMetadata(InetAddressAndPort endpoint) + { + ClusterMetadata metadata = ClusterMetadata.current(); + NodeId nodeId = ClusterMetadataTestHelper.nodeId(2); + NodeAddresses addresses = metadata.directory.getNodeAddresses(nodeId); + Location location = metadata.directory.location(nodeId); + + UntypedResultSet result = executeInternal( + String.format("SELECT * FROM %s.%s WHERE peer = ?", + SYSTEM_KEYSPACE_NAME, LEGACY_PEERS), + endpoint.getAddress()); + UntypedResultSet.Row row = result.one(); + + assertEquals(addresses.broadcastAddress.getAddress(), row.getInetAddress("preferred_ip")); + assertEquals(addresses.nativeAddress.getAddress(), row.getInetAddress("rpc_address")); + assertEquals(location.datacenter, row.getString("data_center")); + assertEquals(location.rack, row.getString("rack")); + assertEquals(nodeId.toUUID(), row.getUUID("host_id")); + assertEquals(metadata.directory.version(nodeId).cassandraVersion.toString(), row.getString("release_version")); + assertEquals(metadata.schema.getVersion(), row.getUUID("schema_version")); + assertEquals(SystemKeyspace.tokensAsSet(metadata.tokenMap.tokens(nodeId)), row.getSet("tokens", UTF8Type.instance)); + } + + private void insertStalePeerEntry(InetAddressAndPort endpoint, String table) + { + if (table.equals(PEERS_V2)) + executeInternal( + String.format("INSERT INTO %s.%s (peer, peer_port, data_center, host_id, rack, release_version, tokens) " + + "VALUES (?, ?, ?, ?, ?, ?, ?)", + SYSTEM_KEYSPACE_NAME, table), + endpoint.getAddress(), endpoint.getPort(), "dc1", UUID.randomUUID(), "rack1", "5.0.0", new HashSet()); + else + executeInternal( + String.format("INSERT INTO %s.%s (peer, data_center, host_id, rack, release_version, tokens) " + + "VALUES (?, ?, ?, ?, ?, ?)", + SYSTEM_KEYSPACE_NAME, table), + endpoint.getAddress(), "dc1", UUID.randomUUID(), "rack1", "5.0.0", new HashSet()); + } + + private void removePeerEntry(InetAddressAndPort endpoint) + { + executeInternal( + String.format("DELETE FROM %s.%s WHERE peer = ? AND peer_port = ?", SYSTEM_KEYSPACE_NAME, PEERS_V2), + endpoint.getAddress(), endpoint.getPort()); + + executeInternal( + String.format("DELETE FROM %s.%s WHERE peer = ?", SYSTEM_KEYSPACE_NAME, LEGACY_PEERS), + endpoint.getAddress()); + } +} \ No newline at end of file