natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add saved Host IDs to TokenMetadata during startup 

Patch by Yifan Cai; reviewed by Sam Tunnicliffe for CASSANDRA-16246
This commit is contained in:
yifan-c 2020-11-05 17:54:11 -08:00 committed by Sam Tunnicliffe
parent 88b18a603c
commit fde640fe52
4 changed files with 68 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

73
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -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<InetAddressAndPort, Token> 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<InetAddressAndPort, Token> loadedTokens = SystemKeyspace.loadTokens();
Map<InetAddressAndPort, UUID> 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<InetAddressAndPort, Token> loadedTokens = SystemKeyspace.loadTokens();
// entry has been mistakenly added, delete it
if (loadedTokens.containsKey(FBUtilities.getBroadcastAddressAndPort()))
SystemKeyspace.removeEndpoint(FBUtilities.getBroadcastAddressAndPort());
Map<InetAddressAndPort, UUID> 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.

9
test/distributed/org/apache/cassandra/distributed/impl/Instance.java
View File

@ -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();

26
test/distributed/org/apache/cassandra/distributed/test/NetworkTopologyTest.java
View File

@ -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")));
}
}
@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<String> 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());
}
}
}