import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
private static Collection<InetAddress> getBatchlogEndpoints(String localDataCenter, ConsistencyLevel consistencyLevel)
throws UnavailableException
{
    TokenMetadata.Topology topology = StorageService.instance.getTokenMetadata().cachedOnlyTokenMap().getTopology();
    Multimap<String, InetAddress> localEndpoints = HashMultimap.create(topology.getDatacenterRacks().get(localDataCenter));
    String localRack = DatabaseDescriptor.getEndpointSnitch().getRack(FBUtilities.getBroadcastAddress());

    Collection<InetAddress> chosenEndpoints = new BatchlogManager.EndpointFilter(localRack, localEndpoints).filter();
    if (chosenEndpoints.isEmpty())
    {
        if (consistencyLevel == ConsistencyLevel.ANY)
            return Collections.singleton(FBUtilities.getBroadcastAddress());

        throw new UnavailableException(ConsistencyLevel.ONE, 1, 0);
    }

    return chosenEndpoints;
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
@Test
public void testGetNeighborsTimesTwo() throws Throwable
{
    TokenMetadata tmd = StorageService.instance.getTokenMetadata();

    // generate rf*2 nodes, and ensure that only neighbors specified by the ARS are returned
    addTokens(2 * Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor());
    AbstractReplicationStrategy ars = Keyspace.open(keyspaceName).getReplicationStrategy();
    Set<InetAddress> expected = new HashSet<InetAddress>();
    for (Range<Token> replicaRange : ars.getAddressRanges().get(FBUtilities.getBroadcastAddress()))
    {
        expected.addAll(ars.getRangeAddresses(tmd.cloneOnlyTokenMap()).get(replicaRange));
    }
    expected.remove(FBUtilities.getBroadcastAddress());
    Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(keyspaceName);
    Set<InetAddress> neighbors = new HashSet<InetAddress>();
    for (Range<Token> range : ranges)
    {
        neighbors.addAll(ActiveRepairService.getNeighbors(keyspaceName, range, null, null));
    }
    assertEquals(expected, neighbors);
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
@Test
public void testGetNeighborsPlusOneInLocalDC() throws Throwable
{
    TokenMetadata tmd = StorageService.instance.getTokenMetadata();
    
    // generate rf+1 nodes, and ensure that all nodes are returned
    Set<InetAddress> expected = addTokens(1 + Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor());
    expected.remove(FBUtilities.getBroadcastAddress());
    // remove remote endpoints
    TokenMetadata.Topology topology = tmd.cloneOnlyTokenMap().getTopology();
    HashSet<InetAddress> localEndpoints = Sets.newHashSet(topology.getDatacenterEndpoints().get(DatabaseDescriptor.getLocalDataCenter()));
    expected = Sets.intersection(expected, localEndpoints);

    Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(keyspaceName);
    Set<InetAddress> neighbors = new HashSet<InetAddress>();
    for (Range<Token> range : ranges)
    {
        neighbors.addAll(ActiveRepairService.getNeighbors(keyspaceName, range, Arrays.asList(DatabaseDescriptor.getLocalDataCenter()), null));
    }
    assertEquals(expected, neighbors);
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
@Test
public void testGetNeighborsTimesTwoInSpecifiedHosts() throws Throwable
{
    TokenMetadata tmd = StorageService.instance.getTokenMetadata();

    // generate rf*2 nodes, and ensure that only neighbors specified by the hosts are returned
    addTokens(2 * Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor());
    AbstractReplicationStrategy ars = Keyspace.open(keyspaceName).getReplicationStrategy();
    List<InetAddress> expected = new ArrayList<>();
    for (Range<Token> replicaRange : ars.getAddressRanges().get(FBUtilities.getBroadcastAddress()))
    {
        expected.addAll(ars.getRangeAddresses(tmd.cloneOnlyTokenMap()).get(replicaRange));
    }

    expected.remove(FBUtilities.getBroadcastAddress());
    Collection<String> hosts = Arrays.asList(FBUtilities.getBroadcastAddress().getCanonicalHostName(),expected.get(0).getCanonicalHostName());

   assertEquals(expected.get(0), ActiveRepairService.getNeighbors(keyspaceName, StorageService.instance.getLocalRanges(keyspaceName).iterator().next(), null, hosts).iterator().next());
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
@Test
public void testRowCacheCleanup() throws Exception
{
    StorageService.instance.initServer(0);
    CacheService.instance.setRowCacheCapacityInMB(1);
    rowCacheLoad(100, Integer.MAX_VALUE, 1000);

    ColumnFamilyStore store = Keyspace.open(KEYSPACE).getColumnFamilyStore(COLUMN_FAMILY);
    assertEquals(CacheService.instance.rowCache.getKeySet().size(), 100);
    store.cleanupCache();
    assertEquals(CacheService.instance.rowCache.getKeySet().size(), 100);
    TokenMetadata tmd = StorageService.instance.getTokenMetadata();
    byte[] tk1, tk2;
    tk1 = "key1000".getBytes();
    tk2 = "key1050".getBytes();
    tmd.updateNormalToken(new BytesToken(tk1), InetAddress.getByName("127.0.0.1"));
    tmd.updateNormalToken(new BytesToken(tk2), InetAddress.getByName("127.0.0.2"));
    store.cleanupCache();
    assertEquals(CacheService.instance.rowCache.getKeySet().size(), 50);
    CacheService.instance.setRowCacheCapacityInMB(0);
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
@Test
public void testRowCacheCleanup() throws Exception
{
    StorageService.instance.initServer(0);
    CacheService.instance.setRowCacheCapacityInMB(1);
    rowCacheLoad(100, Integer.MAX_VALUE, 1000);

    ColumnFamilyStore store = Keyspace.open(KEYSPACE_CACHED).getColumnFamilyStore(CF_CACHED);
    assertEquals(CacheService.instance.rowCache.size(), 100);
    store.cleanupCache();
    assertEquals(CacheService.instance.rowCache.size(), 100);
    TokenMetadata tmd = StorageService.instance.getTokenMetadata();
    byte[] tk1, tk2;
    tk1 = "key1000".getBytes();
    tk2 = "key1050".getBytes();
    tmd.updateNormalToken(new BytesToken(tk1), InetAddress.getByName("127.0.0.1"));
    tmd.updateNormalToken(new BytesToken(tk2), InetAddress.getByName("127.0.0.2"));
    store.cleanupCache();
    assertEquals(50, CacheService.instance.rowCache.size());
    CacheService.instance.setRowCacheCapacityInMB(0);
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
@Test
public void testGetNeighborsTimesTwo() throws Throwable
{
    TokenMetadata tmd = StorageService.instance.getTokenMetadata();

    // generate rf*2 nodes, and ensure that only neighbors specified by the ARS are returned
    addTokens(2 * Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor());
    AbstractReplicationStrategy ars = Keyspace.open(keyspaceName).getReplicationStrategy();
    Set<InetAddress> expected = new HashSet<InetAddress>();
    for (Range<Token> replicaRange : ars.getAddressRanges().get(FBUtilities.getBroadcastAddress()))
    {
        expected.addAll(ars.getRangeAddresses(tmd.cloneOnlyTokenMap()).get(replicaRange));
    }
    expected.remove(FBUtilities.getBroadcastAddress());
    Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(keyspaceName);
    Set<InetAddress> neighbors = new HashSet<InetAddress>();
    for (Range<Token> range : ranges)
    {
        neighbors.addAll(ActiveRepairService.getNeighbors(keyspaceName, range, false));
    }
    assertEquals(expected, neighbors);
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
@Test
public void testGetNeighborsPlusOneInLocalDC() throws Throwable
{
    TokenMetadata tmd = StorageService.instance.getTokenMetadata();
    
    // generate rf+1 nodes, and ensure that all nodes are returned
    Set<InetAddress> expected = addTokens(1 + Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor());
    expected.remove(FBUtilities.getBroadcastAddress());
    // remove remote endpoints
    TokenMetadata.Topology topology = tmd.cloneOnlyTokenMap().getTopology();
    HashSet<InetAddress> localEndpoints = Sets.newHashSet(topology.getDatacenterEndpoints().get(DatabaseDescriptor.getLocalDataCenter()));
    expected = Sets.intersection(expected, localEndpoints);

    Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(keyspaceName);
    Set<InetAddress> neighbors = new HashSet<InetAddress>();
    for (Range<Token> range : ranges)
    {
        neighbors.addAll(ActiveRepairService.getNeighbors(keyspaceName, range, true));
    }
    assertEquals(expected, neighbors);
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
public RangeStreamer(TokenMetadata metadata,
                     Collection<Token> tokens,
                     InetAddress address,
                     String description,
                     boolean useStrictConsistency,
                     IEndpointSnitch snitch,
                     StreamStateStore stateStore)
{
    this.metadata = metadata;
    this.tokens = tokens;
    this.address = address;
    this.description = description;
    this.streamPlan = new StreamPlan(description, true);
    this.useStrictConsistency = useStrictConsistency;
    this.snitch = snitch;
    this.stateStore = stateStore;
    streamPlan.listeners(this.stateStore);
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
public static Collection<Token> allocateTokens(final TokenMetadata tokenMetadata,
                                               final AbstractReplicationStrategy rs,
                                               final InetAddress endpoint,
                                               int numTokens)
{
    StrategyAdapter strategy = getStrategy(tokenMetadata, rs, endpoint);
    Collection<Token> tokens = create(tokenMetadata, strategy).addUnit(endpoint, numTokens);
    tokens = adjustForCrossDatacenterClashes(tokenMetadata, strategy, tokens);

    if (logger.isWarnEnabled())
    {
        logger.warn("Selected tokens {}", tokens);
        SummaryStatistics os = replicatedOwnershipStats(tokenMetadata, rs, endpoint);
        TokenMetadata tokenMetadataCopy = tokenMetadata.cloneOnlyTokenMap();
        tokenMetadataCopy.updateNormalTokens(tokens, endpoint);
        SummaryStatistics ns = replicatedOwnershipStats(tokenMetadataCopy, rs, endpoint);
        logger.warn("Replicated node load in datacentre before allocation " + statToString(os));
        logger.warn("Replicated node load in datacentre after allocation " + statToString(ns));

        // TODO: Is it worth doing the replicated ownership calculation always to be able to raise this alarm?
        if (ns.getStandardDeviation() > os.getStandardDeviation())
            logger.warn("Unexpected growth in standard deviation after allocation.");
    }
    return tokens;
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
static StrategyAdapter getStrategy(final TokenMetadata tokenMetadata, final SimpleStrategy rs, final InetAddress endpoint)
{
    final int replicas = rs.getReplicationFactor();

    return new StrategyAdapter()
    {
        @Override
        public int replicas()
        {
            return replicas;
        }

        @Override
        public Object getGroup(InetAddress unit)
        {
            return unit;
        }

        @Override
        public boolean inAllocationRing(InetAddress other)
        {
            return true;
        }
    };
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
/**
 * if initialtoken was specified, use that (split on comma).
 * otherwise, if allocationKeyspace is specified use the token allocation algorithm to generate suitable tokens
 * else choose num_tokens tokens at random
 */
public static Collection<Token> getBootstrapTokens(final TokenMetadata metadata, InetAddress address) throws ConfigurationException
{
    String allocationKeyspace = DatabaseDescriptor.getAllocateTokensForKeyspace();
    Collection<String> initialTokens = DatabaseDescriptor.getInitialTokens();
    if (initialTokens.size() > 0 && allocationKeyspace != null)
        logger.warn("manually specified tokens override automatic allocation");

    // if user specified tokens, use those
    if (initialTokens.size() > 0)
        return getSpecifiedTokens(metadata, initialTokens);

    int numTokens = DatabaseDescriptor.getNumTokens();
    if (numTokens < 1)
        throw new ConfigurationException("num_tokens must be >= 1");

    if (allocationKeyspace != null)
        return allocateTokens(metadata, address, allocationKeyspace, numTokens);

    if (numTokens == 1)
        logger.warn("Picking random token for a single vnode.  You should probably add more vnodes and/or use the automatic token allocation mechanism.");

    return getRandomTokens(metadata, numTokens);
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
public void testAllocateTokensNetworkStrategy(int rackCount, int replicas) throws UnknownHostException
{
    IEndpointSnitch oldSnitch = DatabaseDescriptor.getEndpointSnitch();
    try
    {
        DatabaseDescriptor.setEndpointSnitch(new RackInferringSnitch());
        int vn = 16;
        String ks = "BootStrapperTestNTSKeyspace" + rackCount + replicas;
        String dc = "1";
        SchemaLoader.createKeyspace(ks, KeyspaceParams.nts(dc, replicas, "15", 15), SchemaLoader.standardCFMD(ks, "Standard1"));
        TokenMetadata tm = new TokenMetadata();
        tm.clearUnsafe();
        for (int i = 0; i < rackCount; ++i)
            generateFakeEndpoints(tm, 10, vn, dc, Integer.toString(i));
        InetAddress addr = InetAddress.getByName("127." + dc + ".0.99");
        allocateTokensForNode(vn, ks, tm, addr);
        // Note: Not matching replication factor in second datacentre, but this should not affect us.
    } finally {
        DatabaseDescriptor.setEndpointSnitch(oldSnitch);
    }
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
@Test
public void testGetNeighborsTimesTwo() throws Throwable
{
    TokenMetadata tmd = StorageService.instance.getTokenMetadata();

    // generate rf*2 nodes, and ensure that only neighbors specified by the ARS are returned
    addTokens(2 * Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor());
    AbstractReplicationStrategy ars = Keyspace.open(KEYSPACE5).getReplicationStrategy();
    Set<InetAddress> expected = new HashSet<>();
    for (Range<Token> replicaRange : ars.getAddressRanges().get(FBUtilities.getBroadcastAddress()))
    {
        expected.addAll(ars.getRangeAddresses(tmd.cloneOnlyTokenMap()).get(replicaRange));
    }
    expected.remove(FBUtilities.getBroadcastAddress());
    Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(KEYSPACE5);
    Set<InetAddress> neighbors = new HashSet<>();
    for (Range<Token> range : ranges)
    {
        neighbors.addAll(ActiveRepairService.getNeighbors(KEYSPACE5, ranges, range, null, null));
    }
    assertEquals(expected, neighbors);
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
@Test
public void testGetNeighborsPlusOneInLocalDC() throws Throwable
{
    TokenMetadata tmd = StorageService.instance.getTokenMetadata();

    // generate rf+1 nodes, and ensure that all nodes are returned
    Set<InetAddress> expected = addTokens(1 + Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor());
    expected.remove(FBUtilities.getBroadcastAddress());
    // remove remote endpoints
    TokenMetadata.Topology topology = tmd.cloneOnlyTokenMap().getTopology();
    HashSet<InetAddress> localEndpoints = Sets.newHashSet(topology.getDatacenterEndpoints().get(DatabaseDescriptor.getLocalDataCenter()));
    expected = Sets.intersection(expected, localEndpoints);

    Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(KEYSPACE5);
    Set<InetAddress> neighbors = new HashSet<>();
    for (Range<Token> range : ranges)
    {
        neighbors.addAll(ActiveRepairService.getNeighbors(KEYSPACE5, ranges, range, Arrays.asList(DatabaseDescriptor.getLocalDataCenter()), null));
    }
    assertEquals(expected, neighbors);
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
@Test
public void testGetNeighborsTimesTwoInSpecifiedHosts() throws Throwable
{
    TokenMetadata tmd = StorageService.instance.getTokenMetadata();

    // generate rf*2 nodes, and ensure that only neighbors specified by the hosts are returned
    addTokens(2 * Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor());
    AbstractReplicationStrategy ars = Keyspace.open(KEYSPACE5).getReplicationStrategy();
    List<InetAddress> expected = new ArrayList<>();
    for (Range<Token> replicaRange : ars.getAddressRanges().get(FBUtilities.getBroadcastAddress()))
    {
        expected.addAll(ars.getRangeAddresses(tmd.cloneOnlyTokenMap()).get(replicaRange));
    }

    expected.remove(FBUtilities.getBroadcastAddress());
    Collection<String> hosts = Arrays.asList(FBUtilities.getBroadcastAddress().getCanonicalHostName(),expected.get(0).getCanonicalHostName());
    Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(KEYSPACE5);

    assertEquals(expected.get(0), ActiveRepairService.getNeighbors(KEYSPACE5, ranges,
                                                                   ranges.iterator().next(),
                                                                   null, hosts).iterator().next());
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
@Test
public void testCleanupWithNewToken() throws ExecutionException, InterruptedException, UnknownHostException
{
    StorageService.instance.getTokenMetadata().clearUnsafe();

    Keyspace keyspace = Keyspace.open(KEYSPACE1);
    ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_STANDARD1);

    // insert data and verify we get it back w/ range query
    fillCF(cfs, "val", LOOPS);

    assertEquals(LOOPS, Util.getAll(Util.cmd(cfs).build()).size());
    TokenMetadata tmd = StorageService.instance.getTokenMetadata();

    byte[] tk1 = new byte[1], tk2 = new byte[1];
    tk1[0] = 2;
    tk2[0] = 1;
    tmd.updateNormalToken(new BytesToken(tk1), InetAddress.getByName("127.0.0.1"));
    tmd.updateNormalToken(new BytesToken(tk2), InetAddress.getByName("127.0.0.2"));
    CompactionManager.instance.performCleanup(cfs, 2);

    assertEquals(0, Util.getAll(Util.cmd(cfs).build()).size());
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
public RangeStreamer(TokenMetadata metadata, Collection<Token> tokens, InetAddress address, String description)
{
    this.metadata = metadata;
    this.tokens = tokens;
    this.address = address;
    this.description = description;
    this.streamPlan = new StreamPlan(description);
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
public BootStrapper(InetAddress address, Collection<Token> tokens, TokenMetadata tmd)
{
    assert address != null;
    assert tokens != null && !tokens.isEmpty();

    this.address = address;
    this.tokens = tokens;
    tokenMetadata = tmd;
}
 

import org.apache.cassandra.locator.TokenMetadata; //导入依赖的package包/类
/**
 * if initialtoken was specified, use that (split on comma).
 * otherwise, if num_tokens == 1, pick a token to assume half the load of the most-loaded node.
 * else choose num_tokens tokens at random
 */
public static Collection<Token> getBootstrapTokens(final TokenMetadata metadata) throws ConfigurationException
{
    Collection<String> initialTokens = DatabaseDescriptor.getInitialTokens();
    // if user specified tokens, use those
    if (initialTokens.size() > 0)
    {
        logger.debug("tokens manually specified as {}",  initialTokens);
        List<Token> tokens = new ArrayList<Token>(initialTokens.size());
        for (String tokenString : initialTokens)
        {
            Token token = StorageService.getPartitioner().getTokenFactory().fromString(tokenString);
            if (metadata.getEndpoint(token) != null)
                throw new ConfigurationException("Bootstrapping to existing token " + tokenString + " is not allowed (decommission/removenode the old node first).");
            tokens.add(token);
        }
        return tokens;
    }

    int numTokens = DatabaseDescriptor.getNumTokens();
    if (numTokens < 1)
        throw new ConfigurationException("num_tokens must be >= 1");

    if (numTokens == 1)
        logger.warn("Picking random token for a single vnode.  You should probably add more vnodes; failing that, you should probably specify the token manually");

    return getRandomTokens(metadata, numTokens);
}