diff --git a/CHANGES.txt b/CHANGES.txt index cb4d5bca9d..cbcc09192c 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0-beta4 + * Add saved Host IDs to TokenMetadata at startup (CASSANDRA-16246) * Ensure that CacheMetrics.requests is picked up by the metric reporter (CASSANDRA-16228) * Add a ratelimiter to snapshot creation and deletion (CASSANDRA-13019) * Produce consistent tombstone for reads to avoid digest mistmatch (CASSANDRA-15369) diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 4a3477cb99..3201d80686 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -637,21 +637,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE MessagingService.instance().listen(); } - public void populateTokenMetadata() - { - if (Boolean.parseBoolean(System.getProperty("cassandra.load_ring_state", "true"))) - { - logger.info("Populating token metadata from system tables"); - Multimap loadedTokens = SystemKeyspace.loadTokens(); - if (!shouldBootstrap()) // if we have not completed bootstrapping, we should not add ourselves as a normal token - loadedTokens.putAll(FBUtilities.getBroadcastAddressAndPort(), SystemKeyspace.getSavedTokens()); - for (InetAddressAndPort ep : loadedTokens.keySet()) - tokenMetadata.updateNormalTokens(loadedTokens.get(ep), ep); - - logger.info("Token metadata: {}", tokenMetadata); - } - } - public synchronized void initServer() throws ConfigurationException { initServer(RING_DELAY); @@ -676,6 +661,14 @@ public class StorageService extends NotificationBroadcasterSupport implements IE throw new AssertionError(e); } + if (Boolean.parseBoolean(System.getProperty("cassandra.load_ring_state", "true"))) + { + logger.info("Loading persisted ring state"); + populatePeerTokenMetadata(); + for (InetAddressAndPort endpoint : tokenMetadata.getAllEndpoints()) + Gossiper.runInGossipStageBlocking(() -> Gossiper.instance.addSavedEndpoint(endpoint)); + } + // daemon threads, like our executors', continue to run while shutdown hooks are invoked drainOnShutdown = NamedThreadFactory.createThread(new WrappedRunnable() { @@ -697,8 +690,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE if (!Boolean.parseBoolean(System.getProperty("cassandra.start_gossip", "true"))) { logger.info("Not starting gossip as requested."); - // load ring state in preparation for starting gossip later - loadRingState(); initialized = true; return; } @@ -740,27 +731,34 @@ public class StorageService extends NotificationBroadcasterSupport implements IE initialized = true; } - private void loadRingState() + public void populateTokenMetadata() { if (Boolean.parseBoolean(System.getProperty("cassandra.load_ring_state", "true"))) { - logger.info("Loading persisted ring state"); - Multimap loadedTokens = SystemKeyspace.loadTokens(); - Map loadedHostIds = SystemKeyspace.loadHostIds(); - for (InetAddressAndPort ep : loadedTokens.keySet()) - { - if (ep.equals(FBUtilities.getBroadcastAddressAndPort())) - { - // entry has been mistakenly added, delete it - SystemKeyspace.removeEndpoint(ep); - } - else - { - if (loadedHostIds.containsKey(ep)) - tokenMetadata.updateHostId(loadedHostIds.get(ep), ep); - Gossiper.runInGossipStageBlocking(() -> Gossiper.instance.addSavedEndpoint(ep)); - } - } + populatePeerTokenMetadata(); + // if we have not completed bootstrapping, we should not add ourselves as a normal token + if (!shouldBootstrap()) + tokenMetadata.updateNormalTokens(SystemKeyspace.getSavedTokens(), FBUtilities.getBroadcastAddressAndPort()); + + logger.info("Token metadata: {}", tokenMetadata); + } + } + + private void populatePeerTokenMetadata() + { + logger.info("Populating token metadata from system tables"); + Multimap loadedTokens = SystemKeyspace.loadTokens(); + + // entry has been mistakenly added, delete it + if (loadedTokens.containsKey(FBUtilities.getBroadcastAddressAndPort())) + SystemKeyspace.removeEndpoint(FBUtilities.getBroadcastAddressAndPort()); + + Map loadedHostIds = SystemKeyspace.loadHostIds(); + for (InetAddressAndPort ep : loadedTokens.keySet()) + { + tokenMetadata.updateNormalTokens(loadedTokens.get(ep), ep); + if (loadedHostIds.containsKey(ep)) + tokenMetadata.updateHostId(loadedHostIds.get(ep), ep); } } @@ -868,11 +866,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE appStates.put(ApplicationState.RPC_ADDRESS, valueFactory.rpcaddress(FBUtilities.getJustBroadcastNativeAddress())); appStates.put(ApplicationState.RELEASE_VERSION, valueFactory.releaseVersion()); - // load the persisted ring state. This used to be done earlier in the init process, - // but now we always perform a shadow round when preparing to join and we have to - // clear endpoint states after doing that. - loadRingState(); - logger.info("Starting up server gossip"); Gossiper.instance.register(this); Gossiper.instance.start(SystemKeyspace.incrementAndGetGeneration(), appStates); // needed for node-ring gathering. diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java index 7ccb59b489..546f31818f 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@ -420,6 +420,10 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance SystemKeyspace.persistLocalMetadata(); SystemKeyspaceMigrator40.migrate(); + // Same order to populate tokenMetadata for the first time, + // see org.apache.cassandra.service.CassandraDaemon.setup + StorageService.instance.populateTokenMetadata(); + try { // load schema from disk @@ -485,6 +489,11 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance } StorageService.instance.ensureTraceKeyspace(); + + // Populate tokenMetadata for the second time, + // see org.apache.cassandra.service.CassandraDaemon.setup + StorageService.instance.populateTokenMetadata(); + SystemKeyspace.finishStartup(); CassandraDaemon.getInstanceForTesting().setupCompleted(); diff --git a/test/distributed/org/apache/cassandra/distributed/test/NetworkTopologyTest.java b/test/distributed/org/apache/cassandra/distributed/test/NetworkTopologyTest.java index a4968c6503..0b256585aa 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/NetworkTopologyTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/NetworkTopologyTest.java @@ -20,6 +20,7 @@ package org.apache.cassandra.distributed.test; import java.io.IOException; import java.util.Collections; +import java.util.List; import java.util.Set; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -27,11 +28,15 @@ import java.util.stream.Stream; import org.junit.Assert; import org.junit.Test; +import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.ICluster; import org.apache.cassandra.distributed.api.IInstance; import org.apache.cassandra.distributed.api.IInvokableInstance; import org.apache.cassandra.distributed.shared.NetworkTopology; +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; + // TODO: this test should be removed after running in-jvm dtests is set up via the shared API repository public class NetworkTopologyTest extends TestBaseImpl { @@ -98,4 +103,23 @@ public class NetworkTopologyTest extends TestBaseImpl { builder().withNodeIdTopology(Collections.singletonMap(2, NetworkTopology.dcAndRack("doomed", "rack"))); } -} \ No newline at end of file + + @Test + public void noWarningForNetworkTopologyStategyConfigOnRestart() throws Exception { + int nodesPerDc = 2; + try (Cluster cluster = builder().withConfig(c -> c.with(GOSSIP, NETWORK)) + .withRacks(2, 1, nodesPerDc) + .start()) { + cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + + " WITH replication = {'class': 'NetworkTopologyStrategy', " + + "'datacenter1' : " + nodesPerDc + ", 'datacenter2' : " + nodesPerDc + " };"); + cluster.get(2).nodetool("flush"); + // Stop node 2 in datacenter 1 + cluster.get(2).shutdown().get(); + // Restart node 2 in datacenter 1 + cluster.get(2).startup(); + List result = cluster.get(2).logs().grep("Ignoring Unrecognized strategy option \\{datacenter2\\}").getResult(); + Assert.assertTrue("Not expected to see the warning about unrecognized option", result.isEmpty()); + } + } +}