test/long/org/apache/cassandra/dht/tokenallocator/NoReplicationTokenAllocatorTest.java - cassandra - Git at Google

 /*
  * 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.dht.tokenallocator;

 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.NavigableMap;
 import java.util.PriorityQueue;
 import java.util.Random;

 import com.google.common.collect.Maps;
 import org.apache.commons.math3.stat.descriptive.SummaryStatistics;
 import org.junit.Test;

 import junit.framework.Assert;
 import org.apache.cassandra.dht.IPartitioner;
 import org.apache.cassandra.dht.Murmur3Partitioner;
 import org.apache.cassandra.dht.RandomPartitioner;
 import org.apache.cassandra.dht.Token;

 public class NoReplicationTokenAllocatorTest extends TokenAllocatorTestBase
 {

     @Test
     public void testNewClusterWithMurmur3Partitioner()
     {
         testNewCluster(new Murmur3Partitioner());
     }

     @Test
     public void testNewClusterWithRandomPartitioner()
     {
         testNewCluster(new RandomPartitioner());
     }

     private void testNewCluster(IPartitioner partitioner)
     {
         for (int perUnitCount = 1; perUnitCount <= MAX_VNODE_COUNT; perUnitCount *= 4)
         {
             testNewCluster(perUnitCount, fixedTokenCount, new NoReplicationStrategy(), partitioner);
             testNewCluster(perUnitCount, fixedTokenCount, new ZeroReplicationStrategy(), partitioner);
         }
     }

     public void testNewCluster(int perUnitCount, TokenCount tc, NoReplicationStrategy rs, IPartitioner partitioner)
     {
         System.out.println("Testing new cluster, target " + perUnitCount + " vnodes, replication " + rs);
         final int targetClusterSize = TARGET_CLUSTER_SIZE;
         NavigableMap<Token, Unit> tokenMap = Maps.newTreeMap();

         NoReplicationTokenAllocator<Unit> t = new NoReplicationTokenAllocator<Unit>(tokenMap, rs, partitioner);
         grow(t, targetClusterSize * 2 / 5, tc, perUnitCount, false);
         grow(t, targetClusterSize, tc, perUnitCount, true);
         System.out.println();
     }

     @Test
     public void testExistingClusterWithMurmur3Partitioner()
     {
         testExistingCluster(new Murmur3Partitioner());
     }

     @Test
     public void testExistingClusterWithRandomPartitioner()
     {
         testExistingCluster(new RandomPartitioner());
     }

     private void testExistingCluster(IPartitioner partitioner)
     {
         for (int perUnitCount = 1; perUnitCount <= MAX_VNODE_COUNT; perUnitCount *= 4)
         {
             testExistingCluster(perUnitCount, fixedTokenCount, new NoReplicationStrategy(), partitioner);
             testExistingCluster(perUnitCount, fixedTokenCount, new ZeroReplicationStrategy(), partitioner);
         }
     }

     public NoReplicationTokenAllocator<Unit> randomWithTokenAllocator(NavigableMap<Token, Unit> map, NoReplicationStrategy rs,
                                                                       int unitCount, TokenCount tc, int perUnitCount,
                                                                       IPartitioner partitioner)
     {
         super.random(map, rs, unitCount, tc, perUnitCount, partitioner);
         NoReplicationTokenAllocator<Unit> t = new NoReplicationTokenAllocator<Unit>(map, rs, partitioner);
         t.createTokenInfos();
         return t;
     }

     public void testExistingCluster(int perUnitCount, TokenCount tc, NoReplicationStrategy rs, IPartitioner partitioner)
     {
         System.out.println("Testing existing cluster, target " + perUnitCount + " vnodes, replication " + rs);
         final int targetClusterSize = TARGET_CLUSTER_SIZE;
         NavigableMap<Token, Unit> tokenMap = Maps.newTreeMap();
         NoReplicationTokenAllocator<Unit> t = randomWithTokenAllocator(tokenMap, rs, targetClusterSize / 2, tc, perUnitCount, partitioner);
         updateSummaryBeforeGrow(t);

         grow(t, targetClusterSize * 9 / 10, tc, perUnitCount, false);
         grow(t, targetClusterSize, tc, perUnitCount, true);
         loseAndReplace(t, targetClusterSize / 10, tc, perUnitCount, partitioner);
         System.out.println();
     }

     private void loseAndReplace(NoReplicationTokenAllocator<Unit> t, int howMany,
                                 TokenCount tc, int perUnitCount, IPartitioner partitioner)
     {
         int fullCount = t.sortedUnits.size();
         System.out.format("Losing %d units. ", howMany);
         for (int i = 0; i < howMany; ++i)
         {
             Unit u = t.unitFor(partitioner.getRandomToken(seededRand));
             t.removeUnit(u);
         }
         // Grow half without verifying.
         grow(t, (t.sortedUnits.size() + fullCount * 3) / 4, tc, perUnitCount, false);
         // Metrics should be back to normal by now. Check that they remain so.
         grow(t, fullCount, tc, perUnitCount, true);
     }

     private void updateSummaryBeforeGrow(NoReplicationTokenAllocator<Unit> t)
     {
         Summary su = new Summary();
         Summary st = new Summary();
         System.out.println("Before growing cluster: ");
         updateSummary(t, su, st, true);
     }

     private void grow(NoReplicationTokenAllocator<Unit> t, int targetClusterSize, TokenCount tc, int perUnitCount, boolean verifyMetrics)
     {
         int size = t.sortedUnits.size();
         Summary su = new Summary();
         Summary st = new Summary();
         Random rand = new Random(targetClusterSize + perUnitCount);
         TestReplicationStrategy strategy = (TestReplicationStrategy) t.strategy;
         if (size < targetClusterSize)
         {
             System.out.format("Adding %d unit(s) using %s...", targetClusterSize - size, t.toString());
             long time = System.currentTimeMillis();

             while (size < targetClusterSize)
             {
                 int num_tokens = tc.tokenCount(perUnitCount, rand);
                 Unit unit = new Unit();
                 t.addUnit(unit, num_tokens);
                 ++size;
                 if (verifyMetrics)
                     updateSummary(t, su, st, false);
             }
             System.out.format(" Done in %.3fs\n", (System.currentTimeMillis() - time) / 1000.0);

             if (verifyMetrics)
             {
                 updateSummary(t, su, st, true);
                 double maxExpected = 1.0 + tc.spreadExpectation() * strategy.spreadExpectation() / perUnitCount;
                 if (su.max > maxExpected)
                 {
                     Assert.fail(String.format("Expected max unit size below %.4f, was %.4f", maxExpected, su.max));
                 }
             }
         }
     }

     private void updateSummary(NoReplicationTokenAllocator<Unit> t, Summary su, Summary st, boolean print)
     {
         int size = t.sortedTokens.size();

         SummaryStatistics unitStat = new SummaryStatistics();
         for (TokenAllocatorBase.Weighted<TokenAllocatorBase.UnitInfo> wu : t.sortedUnits)
         {
             unitStat.addValue(wu.weight * size / t.tokensInUnits.get(wu.value.unit).size());
         }
         su.update(unitStat);

         SummaryStatistics tokenStat = new SummaryStatistics();
         for (PriorityQueue<TokenAllocatorBase.Weighted<TokenAllocatorBase.TokenInfo>> tokens : t.tokensInUnits.values())
         {
             for (TokenAllocatorBase.Weighted<TokenAllocatorBase.TokenInfo> token : tokens)
             {
                 tokenStat.addValue(token.weight);
             }
         }
         st.update(tokenStat);

         if (print)
         {
             System.out.format("Size %d(%d)   \tunit %s  token %s   %s\n",
                               t.sortedUnits.size(), size,
                               mms(unitStat),
                               mms(tokenStat),
                               t.strategy);
             System.out.format("Worst intermediate unit\t%s  token %s\n", su, st);
         }
     }

     static class NoReplicationStrategy implements TestReplicationStrategy
     {
         public List<Unit> getReplicas(Token token, NavigableMap<Token, Unit> sortedTokens)
         {
             return Collections.singletonList(sortedTokens.ceilingEntry(token).getValue());
         }

         public Token replicationStart(Token token, Unit unit, NavigableMap<Token, Unit> sortedTokens)
         {
             return sortedTokens.lowerKey(token);
         }

         public String toString()
         {
             return "No replication";
         }

         public void addUnit(Unit n)
         {
         }

         public void removeUnit(Unit n)
         {
         }

         public int replicas()
         {
             return 1;
         }

         public Object getGroup(Unit unit)
         {
             return unit;
         }

         public double spreadExpectation()
         {
             return 1;
         }
     }

     static class ZeroReplicationStrategy extends NoReplicationStrategy
     {
         public int replicas()
         {
             return 0;
         }
     }
 }
	/*
	* 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.dht.tokenallocator;

	import java.util.Collections;
	import java.util.List;
	import java.util.Map;
	import java.util.NavigableMap;
	import java.util.PriorityQueue;
	import java.util.Random;

	import com.google.common.collect.Maps;
	import org.apache.commons.math3.stat.descriptive.SummaryStatistics;
	import org.junit.Test;

	import junit.framework.Assert;
	import org.apache.cassandra.dht.IPartitioner;
	import org.apache.cassandra.dht.Murmur3Partitioner;
	import org.apache.cassandra.dht.RandomPartitioner;
	import org.apache.cassandra.dht.Token;

	public class NoReplicationTokenAllocatorTest extends TokenAllocatorTestBase
	{

	@Test
	public void testNewClusterWithMurmur3Partitioner()
	{
	testNewCluster(new Murmur3Partitioner());
	}

	@Test
	public void testNewClusterWithRandomPartitioner()
	{
	testNewCluster(new RandomPartitioner());
	}

	private void testNewCluster(IPartitioner partitioner)
	{
	for (int perUnitCount = 1; perUnitCount <= MAX_VNODE_COUNT; perUnitCount *= 4)
	{
	testNewCluster(perUnitCount, fixedTokenCount, new NoReplicationStrategy(), partitioner);
	testNewCluster(perUnitCount, fixedTokenCount, new ZeroReplicationStrategy(), partitioner);
	}
	}

	public void testNewCluster(int perUnitCount, TokenCount tc, NoReplicationStrategy rs, IPartitioner partitioner)
	{
	System.out.println("Testing new cluster, target " + perUnitCount + " vnodes, replication " + rs);
	final int targetClusterSize = TARGET_CLUSTER_SIZE;
	NavigableMap<Token, Unit> tokenMap = Maps.newTreeMap();

	NoReplicationTokenAllocator<Unit> t = new NoReplicationTokenAllocator<Unit>(tokenMap, rs, partitioner);
	grow(t, targetClusterSize * 2 / 5, tc, perUnitCount, false);
	grow(t, targetClusterSize, tc, perUnitCount, true);
	System.out.println();
	}

	@Test
	public void testExistingClusterWithMurmur3Partitioner()
	{
	testExistingCluster(new Murmur3Partitioner());
	}

	@Test
	public void testExistingClusterWithRandomPartitioner()
	{
	testExistingCluster(new RandomPartitioner());
	}

	private void testExistingCluster(IPartitioner partitioner)
	{
	for (int perUnitCount = 1; perUnitCount <= MAX_VNODE_COUNT; perUnitCount *= 4)
	{
	testExistingCluster(perUnitCount, fixedTokenCount, new NoReplicationStrategy(), partitioner);
	testExistingCluster(perUnitCount, fixedTokenCount, new ZeroReplicationStrategy(), partitioner);
	}
	}

	public NoReplicationTokenAllocator<Unit> randomWithTokenAllocator(NavigableMap<Token, Unit> map, NoReplicationStrategy rs,
	int unitCount, TokenCount tc, int perUnitCount,
	IPartitioner partitioner)
	{
	super.random(map, rs, unitCount, tc, perUnitCount, partitioner);
	NoReplicationTokenAllocator<Unit> t = new NoReplicationTokenAllocator<Unit>(map, rs, partitioner);
	t.createTokenInfos();
	return t;
	}

	public void testExistingCluster(int perUnitCount, TokenCount tc, NoReplicationStrategy rs, IPartitioner partitioner)
	{
	System.out.println("Testing existing cluster, target " + perUnitCount + " vnodes, replication " + rs);
	final int targetClusterSize = TARGET_CLUSTER_SIZE;
	NavigableMap<Token, Unit> tokenMap = Maps.newTreeMap();
	NoReplicationTokenAllocator<Unit> t = randomWithTokenAllocator(tokenMap, rs, targetClusterSize / 2, tc, perUnitCount, partitioner);
	updateSummaryBeforeGrow(t);

	grow(t, targetClusterSize * 9 / 10, tc, perUnitCount, false);
	grow(t, targetClusterSize, tc, perUnitCount, true);
	loseAndReplace(t, targetClusterSize / 10, tc, perUnitCount, partitioner);
	System.out.println();
	}

	private void loseAndReplace(NoReplicationTokenAllocator<Unit> t, int howMany,
	TokenCount tc, int perUnitCount, IPartitioner partitioner)
	{
	int fullCount = t.sortedUnits.size();
	System.out.format("Losing %d units. ", howMany);
	for (int i = 0; i < howMany; ++i)
	{
	Unit u = t.unitFor(partitioner.getRandomToken(seededRand));
	t.removeUnit(u);
	}
	// Grow half without verifying.
	grow(t, (t.sortedUnits.size() + fullCount * 3) / 4, tc, perUnitCount, false);
	// Metrics should be back to normal by now. Check that they remain so.
	grow(t, fullCount, tc, perUnitCount, true);
	}

	private void updateSummaryBeforeGrow(NoReplicationTokenAllocator<Unit> t)
	{
	Summary su = new Summary();
	Summary st = new Summary();
	System.out.println("Before growing cluster: ");
	updateSummary(t, su, st, true);
	}

	private void grow(NoReplicationTokenAllocator<Unit> t, int targetClusterSize, TokenCount tc, int perUnitCount, boolean verifyMetrics)
	{
	int size = t.sortedUnits.size();
	Summary su = new Summary();
	Summary st = new Summary();
	Random rand = new Random(targetClusterSize + perUnitCount);
	TestReplicationStrategy strategy = (TestReplicationStrategy) t.strategy;
	if (size < targetClusterSize)
	{
	System.out.format("Adding %d unit(s) using %s...", targetClusterSize - size, t.toString());
	long time = System.currentTimeMillis();

	while (size < targetClusterSize)
	{
	int num_tokens = tc.tokenCount(perUnitCount, rand);
	Unit unit = new Unit();
	t.addUnit(unit, num_tokens);
	++size;
	if (verifyMetrics)
	updateSummary(t, su, st, false);
	}
	System.out.format(" Done in %.3fs\n", (System.currentTimeMillis() - time) / 1000.0);

	if (verifyMetrics)
	{
	updateSummary(t, su, st, true);
	double maxExpected = 1.0 + tc.spreadExpectation() * strategy.spreadExpectation() / perUnitCount;
	if (su.max > maxExpected)
	{
	Assert.fail(String.format("Expected max unit size below %.4f, was %.4f", maxExpected, su.max));
	}
	}
	}
	}

	private void updateSummary(NoReplicationTokenAllocator<Unit> t, Summary su, Summary st, boolean print)
	{
	int size = t.sortedTokens.size();

	SummaryStatistics unitStat = new SummaryStatistics();
	for (TokenAllocatorBase.Weighted<TokenAllocatorBase.UnitInfo> wu : t.sortedUnits)
	{
	unitStat.addValue(wu.weight * size / t.tokensInUnits.get(wu.value.unit).size());
	}
	su.update(unitStat);

	SummaryStatistics tokenStat = new SummaryStatistics();
	for (PriorityQueue<TokenAllocatorBase.Weighted<TokenAllocatorBase.TokenInfo>> tokens : t.tokensInUnits.values())
	{
	for (TokenAllocatorBase.Weighted<TokenAllocatorBase.TokenInfo> token : tokens)
	{
	tokenStat.addValue(token.weight);
	}
	}
	st.update(tokenStat);

	if (print)
	{
	System.out.format("Size %d(%d) \tunit %s token %s %s\n",
	t.sortedUnits.size(), size,
	mms(unitStat),
	mms(tokenStat),
	t.strategy);
	System.out.format("Worst intermediate unit\t%s token %s\n", su, st);
	}
	}

	static class NoReplicationStrategy implements TestReplicationStrategy
	{
	public List<Unit> getReplicas(Token token, NavigableMap<Token, Unit> sortedTokens)
	{
	return Collections.singletonList(sortedTokens.ceilingEntry(token).getValue());
	}

	public Token replicationStart(Token token, Unit unit, NavigableMap<Token, Unit> sortedTokens)
	{
	return sortedTokens.lowerKey(token);
	}

	public String toString()
	{
	return "No replication";
	}

	public void addUnit(Unit n)
	{
	}

	public void removeUnit(Unit n)
	{
	}

	public int replicas()
	{
	return 1;
	}

	public Object getGroup(Unit unit)
	{
	return unit;
	}

	public double spreadExpectation()
	{
	return 1;
	}
	}

	static class ZeroReplicationStrategy extends NoReplicationStrategy
	{
	public int replicas()
	{
	return 0;
	}
	}
	}