diff --git a/src/java/org/apache/cassandra/db/virtual/PeersTable.java b/src/java/org/apache/cassandra/db/virtual/PeersTable.java index 5b011de604..3ce979c69a 100644 --- a/src/java/org/apache/cassandra/db/virtual/PeersTable.java +++ b/src/java/org/apache/cassandra/db/virtual/PeersTable.java @@ -156,9 +156,7 @@ public class PeersTable extends AbstractVirtualTable if (next.directory.peerState(nodeId) == null || next.directory.peerState(nodeId) == NodeState.LEFT) { NodeAddresses addresses = prev.directory.getNodeAddresses(nodeId); - logger.debug("Purging {} from system.peers_v2 table", addresses); - QueryProcessor.executeInternal(String.format(peers_delete_query, SYSTEM_KEYSPACE_NAME, PEERS_V2), addresses.broadcastAddress.getAddress(), addresses.broadcastAddress.getPort()); - QueryProcessor.executeInternal(String.format(legacy_peers_delete_query, SYSTEM_KEYSPACE_NAME, LEGACY_PEERS), addresses.broadcastAddress.getAddress()); + removeFromLegacyPeerTable(addresses.broadcastAddress); } else if (NodeState.isPreJoin(next.directory.peerState(nodeId))) { @@ -197,4 +195,11 @@ public class PeersTable extends AbstractVirtualTable tokens); } } + + public static void removeFromLegacyPeerTable(InetAddressAndPort addr) + { + logger.debug("Purging {} from system.peers_v2 table", addr); + QueryProcessor.executeInternal(String.format(peers_delete_query, SYSTEM_KEYSPACE_NAME, PEERS_V2), addr.getAddress(), addr.getPort()); + QueryProcessor.executeInternal(String.format(legacy_peers_delete_query, SYSTEM_KEYSPACE_NAME, LEGACY_PEERS), addr.getAddress()); + } } \ No newline at end of file diff --git a/src/java/org/apache/cassandra/tcm/Startup.java b/src/java/org/apache/cassandra/tcm/Startup.java index dfe4df8399..810235450b 100644 --- a/src/java/org/apache/cassandra/tcm/Startup.java +++ b/src/java/org/apache/cassandra/tcm/Startup.java @@ -158,8 +158,10 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort; UUID currentHostId = SystemKeyspace.getLocalHostId(); if (nodeId != null && !Objects.equals(nodeId.toUUID(), currentHostId)) { - logger.info("NodeId is wrong, updating from {} to {}", currentHostId, nodeId.toUUID()); - SystemKeyspace.setLocalHostId(nodeId.toUUID()); + String error = String.format("NodeId does not match locally set one. Check for the IP address collision: %s vs %s %s", + currentHostId, nodeId.toUUID(), FBUtilities.getBroadcastAddressAndPort()); + logger.error(error); + throw new IllegalStateException(error); } } diff --git a/src/java/org/apache/cassandra/tcm/listeners/LegacyStateListener.java b/src/java/org/apache/cassandra/tcm/listeners/LegacyStateListener.java index c2199040d3..bdbf4e5a23 100644 --- a/src/java/org/apache/cassandra/tcm/listeners/LegacyStateListener.java +++ b/src/java/org/apache/cassandra/tcm/listeners/LegacyStateListener.java @@ -63,7 +63,9 @@ public class LegacyStateListener implements ChangeListener.Async next.tokenMap.lastModified().equals(prev.tokenMap.lastModified())) return; - Set removed = Sets.difference(prev.directory.peerIds(), next.directory.peerIds()); + Set removedAddr = Sets.difference(new HashSet<>(prev.directory.allAddresses()), + new HashSet<>(next.directory.allAddresses())); + Set changed = new HashSet<>(); for (NodeId node : next.directory.peerIds()) { @@ -71,10 +73,10 @@ public class LegacyStateListener implements ChangeListener.Async changed.add(node); } - for (NodeId remove : removed) + for (InetAddressAndPort remove : removedAddr) { - GossipHelper.evictFromMembership(prev.directory.endpoint(remove)); - PeersTable.updateLegacyPeerTable(remove, prev, next); + GossipHelper.evictFromMembership(remove); + PeersTable.removeFromLegacyPeerTable(remove); } for (NodeId change : changed) diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/BounceResetHostIdTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/BounceResetHostIdTest.java index 0411c41617..e593a71b68 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/log/BounceResetHostIdTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/log/BounceResetHostIdTest.java @@ -18,35 +18,120 @@ package org.apache.cassandra.distributed.test.log; -import java.util.UUID; +import java.net.InetAddress; +import java.util.Arrays; +import java.util.Comparator; +import java.util.concurrent.TimeUnit; +import org.junit.Assert; import org.junit.Test; -import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.shared.AssertUtils; +import org.apache.cassandra.distributed.shared.ClusterUtils; import org.apache.cassandra.distributed.test.TestBaseImpl; import org.apache.cassandra.tcm.membership.NodeId; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; +import static org.apache.cassandra.distributed.shared.AssertUtils.row; +import static org.junit.Assert.fail; public class BounceResetHostIdTest extends TestBaseImpl { @Test - public void bounceTest() throws Exception + public void swapIpsTest() throws Exception { - try (Cluster cluster = init(builder().withNodes(1) - .start())) + try (Cluster cluster = builder().withNodes(3) + .withConfig(c -> c.with(Feature.GOSSIP, Feature.NATIVE_PROTOCOL) + // disable DistributedTestSnitch as it tries to query before we setup + .set("endpoint_snitch", "org.apache.cassandra.locator.SimpleSnitch")) + .createWithoutStarting()) { - String wrongId = UUID.randomUUID().toString(); - cluster.get(1).runOnInstance(() -> { - SystemKeyspace.setLocalHostId(UUID.fromString(wrongId)); - assertFalse(NodeId.isValidNodeId(SystemKeyspace.getLocalHostId())); - }); - cluster.get(1).shutdown().get(); - cluster.get(1).startup(); - cluster.get(1).logs().watchFor("NodeId is wrong, updating from "+wrongId+" to "+(new NodeId(1).toUUID())); - cluster.get(1).runOnInstance(() -> assertTrue(NodeId.isValidNodeId(SystemKeyspace.getLocalHostId()))); + // This test relies on node IDs being in the same order as IP addresses + for (int i = 1; i <= 3; i++) + cluster.get(i).startup(); + + cluster.get(2).shutdown().get(); + ClusterUtils.updateAddress(cluster.get(2), "127.0.0.4"); + cluster.get(2).startup(); + + cluster.get(3).shutdown().get(); + ClusterUtils.updateAddress(cluster.get(3), "127.0.0.2"); + cluster.get(3).startup(); + + cluster.get(2).shutdown().get(); + ClusterUtils.updateAddress(cluster.get(2), "127.0.0.3"); + cluster.get(2).startup(); + + ClusterUtils.waitForCMSToQuiesce(cluster, cluster.get(1)); + + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30); + while (true) + { + try + { + AssertUtils.assertRows(sortHelper(cluster.coordinator(2).execute("select peer, host_id from system.peers_v2", ConsistencyLevel.QUORUM)), + rows(row(InetAddress.getByName("127.0.0.1"), new NodeId(1).toUUID()), + row(InetAddress.getByName("127.0.0.2"), new NodeId(3).toUUID()) + )); + AssertUtils.assertRows(sortHelper(cluster.coordinator(3).execute("select peer, host_id from system.peers_v2", ConsistencyLevel.QUORUM)), + rows(row(InetAddress.getByName("127.0.0.1"), new NodeId(1).toUUID()), + row(InetAddress.getByName("127.0.0.3"), new NodeId(2).toUUID()) + + )); + return; + } + catch (AssertionError t) + { + // If we are past the deadline, throw; allow to retry otherwise + if (System.nanoTime() > deadline) + throw t; + } + } } } + + @Test + public void swapIpsDirectlyTest() throws Exception + { + try (Cluster cluster = builder().withNodes(3) + .withConfig(c -> c.with(Feature.GOSSIP, Feature.NATIVE_PROTOCOL) + // disable DistributedTestSnitch as it tries to query before we setup + .set("endpoint_snitch", "org.apache.cassandra.locator.SimpleSnitch")) + .createWithoutStarting()) + { + // This test relies on node IDs being in the same order as IP addresses + for (int i = 1; i <= 3; i++) + cluster.get(i).startup(); + + cluster.get(2).shutdown().get(); + cluster.get(3).shutdown().get(); + ClusterUtils.updateAddress(cluster.get(2), "127.0.0.3"); + ClusterUtils.updateAddress(cluster.get(3), "127.0.0.2"); + try + { + cluster.get(2).startup(); + fail("Should not have been able to start"); + } + catch (Throwable t) + { + Assert.assertTrue(t.getMessage().contains("NodeId does not match locally set one")); + } + try + { + cluster.get(3).startup(); + fail("Should not have been able to start"); + } + catch (Throwable t) + { + Assert.assertTrue(t.getMessage().contains("NodeId does not match locally set one")); + } + } + } + public static Object[][] sortHelper(Object[][] rows) + { + Arrays.sort(rows, Comparator.comparing(r -> ((InetAddress)r[0]).getHostAddress())); + return rows; + } }