DefaultDistributionPolicy implements IDeployPolicy { public Map<String, List<String>> createDistributionPlan(final Map<String, List<String>> currentShard2NodesMap, final Map<String, List<String>> currentNode2ShardsMap, List<String> aliveNodes, final int replicationLevel) { if (aliveNodes.size() == 0) { throw new IllegalArgumentException("no alive nodes to distribute to"); } Set<String> shards = currentShard2NodesMap.keySet(); for (String shard : shards) { Set<String> assignedNodes = new HashSet<String>(replicationLevel); int neededDeployments = replicationLevel - countValues(currentShard2NodesMap, shard); assignedNodes.addAll(currentShard2NodesMap.get(shard)); sortAfterFreeCapacity(aliveNodes, currentNode2ShardsMap); CircularList<String> roundRobinNodes = new CircularList<String>(aliveNodes); neededDeployments = chooseNewNodes(currentNode2ShardsMap, roundRobinNodes, shard, assignedNodes, neededDeployments); if (neededDeployments > 0) { LOG.warn("cannot replicate shard '" + shard + "' " + replicationLevel + " times, cause only " + roundRobinNodes.size() + " nodes connected"); } else if (neededDeployments < 0) { LOG.info("found shard '" + shard + "' over-replicated"); removeOverreplicatedShards(currentShard2NodesMap, currentNode2ShardsMap, shard, neededDeployments); } } return currentNode2ShardsMap; }  Map<String, List<String>> createDistributionPlan(final Map<String, List<String>> currentShard2NodesMap,
          final Map<String, List<String>> currentNode2ShardsMap, List<String> aliveNodes, final int replicationLevel);  }
@Test public void testInitialDistribution() throws Exception { int replicationLevel = 2; List<String> nodes = createNodes("node1", "node2", "node3"); Set<String> shards = createShards("shard1", "shard2"); Map<String, List<String>> node2ShardsMap = _distributionPolicy.createDistributionPlan(_currentShard2NodesMap, _currentNode2ShardsMap, nodes, replicationLevel); assertEquals(nodes.size(), node2ShardsMap.size()); assertSufficientDistribution(replicationLevel, nodes, shards, node2ShardsMap); }
@Test public void testEqualDistributionOnMultipleSequentialDeploys() throws Exception { int replicationLevel = 1; List<String> nodes = createNodes("node1", "node2", "node3", "node4"); createShards("shard1", "shard2"); Map<String, List<String>> node2ShardsMap = _distributionPolicy.createDistributionPlan(_currentShard2NodesMap, _currentNode2ShardsMap, nodes, replicationLevel); System.out.println(node2ShardsMap); _currentShard2NodesMap.clear(); createShards("shard3", "shard4"); node2ShardsMap = _distributionPolicy.createDistributionPlan(_currentShard2NodesMap, _currentNode2ShardsMap, nodes, replicationLevel); for (String node : node2ShardsMap.keySet()) { assertEquals("shards are not equally distributed: " + node2ShardsMap, 1, node2ShardsMap.get(node).size()); } System.out.println(node2ShardsMap); }
@Test public void testInitialDistribution_TooLessNodes() throws Exception { List<String> nodes = createNodes("node1"); Set<String> shards = createShards("shard1", "shard2"); Map<String, List<String>> node2ShardsMap = _distributionPolicy.createDistributionPlan(_currentShard2NodesMap, _currentNode2ShardsMap, nodes, 3); assertEquals(nodes.size(), node2ShardsMap.size()); assertEquals(shards.size(), node2ShardsMap.get("node1").size()); }
@Test public void testUnderReplicatedDistribution() throws Exception { int replicationLevel = 3; List<String> nodes = createNodes("node1", "node2", "node3"); Set<String> shards = createShards("shard1", "shard2", "shard3"); addMapping("shard1", "node1", "node2", "node3"); addMapping("shard2", "node1"); Map<String, List<String>> node2ShardsMap = _distributionPolicy.createDistributionPlan(_currentShard2NodesMap, _currentNode2ShardsMap, nodes, replicationLevel); assertEquals(nodes.size(), node2ShardsMap.size()); assertSufficientDistribution(replicationLevel, nodes, shards, node2ShardsMap); }
@Test public void testOverReplicatedDistribution() throws Exception { int replicationLevel = 2; List<String> nodes = createNodes("node1", "node2", "node3", "node4"); Set<String> shards = createShards("shard1", "shard2"); addMapping("shard1", "node1", "node2", "node3", "node4"); addMapping("shard2", "node1", "node2"); Map<String, List<String>> node2ShardsMap = _distributionPolicy.createDistributionPlan(_currentShard2NodesMap, _currentNode2ShardsMap, nodes, replicationLevel); assertEquals(nodes.size(), node2ShardsMap.size()); assertSufficientDistribution(replicationLevel, nodes, shards, node2ShardsMap); }