HostAwareDistributionPolicy 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)); List<String> sortedNodes = sortNodesByCapacityHostAware(aliveNodes, shard, currentNode2ShardsMap); CircularList<String> roundRobinNodes = new CircularList<String>(sortedNodes); 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 = 3; List<String> nodes = createNodes("node1:20001", "node1:20002", "node2:20001", "node2:20002", "node3:20001", "node3:20002"); Set<String> shards = createShards("shard1", "shard2", "shard3", "shard4"); Map<String, List<String>> node2ShardsMap = _distributionPolicy.createDistributionPlan(_currentShard2NodesMap, _currentNode2ShardsMap, nodes, replicationLevel); assertEquals(nodes.size(), node2ShardsMap.size()); assertSufficientDistribution(replicationLevel, nodes, shards, node2ShardsMap); assertTrue( Collections.disjoint(node2ShardsMap.get("node1:20001"), node2ShardsMap.get("node1:20002"))); assertTrue( Collections.disjoint(node2ShardsMap.get("node2:20001"), node2ShardsMap.get("node2:20002"))); assertTrue( Collections.disjoint(node2ShardsMap.get("node3:20001"), node2ShardsMap.get("node3:20002"))); }
@Test public void testThreeRoundsDistribution() throws Exception { int replicationLevel = 1; List<String> nodes = createNodes("node1:20001", "node1:20002", "node2:20001", "node2:20002", "node3:20001", "node3:20002"); Set<String> shards = createShards("shard1", "shard2", "shard3", "shard4"); Map<String, List<String>> firstNode2ShardsMap = _distributionPolicy.createDistributionPlan(_currentShard2NodesMap, _currentNode2ShardsMap, nodes, replicationLevel); assertEquals(nodes.size(), firstNode2ShardsMap.size()); updateShardToNodesMap(firstNode2ShardsMap); replicationLevel = 2; Map<String, List<String>> secondNode2ShardsMap = _distributionPolicy.createDistributionPlan(_currentShard2NodesMap, firstNode2ShardsMap, nodes, replicationLevel); assertEquals(nodes.size(), secondNode2ShardsMap.size()); assertSufficientDistribution(replicationLevel, nodes, shards, secondNode2ShardsMap); updateShardToNodesMap(secondNode2ShardsMap); replicationLevel = 3; Map<String, List<String>> thirdNode2ShardsMap = _distributionPolicy.createDistributionPlan(_currentShard2NodesMap, secondNode2ShardsMap, nodes, replicationLevel); assertEquals(nodes.size(), thirdNode2ShardsMap.size()); assertSufficientDistribution(replicationLevel, nodes, shards, thirdNode2ShardsMap); assertTrue( Collections.disjoint(thirdNode2ShardsMap.get("node1:20001"), thirdNode2ShardsMap.get("node1:20002"))); assertTrue( Collections.disjoint(thirdNode2ShardsMap.get("node2:20001"), thirdNode2ShardsMap.get("node2:20002"))); assertTrue( Collections.disjoint(thirdNode2ShardsMap.get("node3:20001"), thirdNode2ShardsMap.get("node3:20002"))); }
@Test public void testTypicalDeployedDistribution() throws Exception { int replicationLevel = 2; List<String> nodes = createNodesByHosts(4, 6); createNumShards(46); Map<String, List<String>> node2ShardsMap = _distributionPolicy.createDistributionPlan(_currentShard2NodesMap, _currentNode2ShardsMap, nodes, replicationLevel); assertEquals(nodes.size(), node2ShardsMap.size()); assertDisjointNodesOnEachHost(4, 6, node2ShardsMap); }
@Test public void testEqualDistributionOnMultipleSequentialDeploys() throws Exception { int replicationLevel = 1; List<String> nodes = createNodes("node1:20000", "node2:20000", "node3:20000", "node4:20000"); 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:20000"); 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:20000").size()); }
@Test public void testUnderReplicatedDistribution() throws Exception { int replicationLevel = 3; List<String> nodes = createNodes("node1:20000", "node2:20000", "node3:20000"); Set<String> shards = createShards("shard1", "shard2", "shard3"); addMapping("shard1", "node1:20000", "node2:20000", "node3:20000"); addMapping("shard2", "node1:20000"); 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:20000", "node2:20000", "node3:20000", "node4:20000"); Set<String> shards = createShards("shard1", "shard2"); addMapping("shard1", "node1:20000", "node2:20000", "node3:20000", "node4:20000"); addMapping("shard2", "node1:20000", "node2:20000"); Map<String, List<String>> node2ShardsMap = _distributionPolicy.createDistributionPlan(_currentShard2NodesMap, _currentNode2ShardsMap, nodes, replicationLevel); assertEquals(nodes.size(), node2ShardsMap.size()); assertSufficientDistribution(replicationLevel, nodes, shards, node2ShardsMap); }