LowestShardCountDistributionPolicy 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"); } if (LOG.isDebugEnabled()) { LOG.debug(String.format("Live nodes are %d, %s", aliveNodes.size(), aliveNodes)); LOG.debug(String.format("Current Shard 2 Nodes Map, %s", currentShard2NodesMap)); LOG.debug(String.format("Current Nodes 2 Shard Map, %s", currentNode2ShardsMap)); } Set<String> shards = currentShard2NodesMap.keySet(); TreeSet<OrderedBucket> orderedNodes = produceOrdedNodeList(currentNode2ShardsMap, aliveNodes); for (String shard : shards) { Set<String> assignedNodes = new HashSet<String>(replicationLevel); int neededDeployments = replicationLevel - countValues(currentShard2NodesMap, shard); assignedNodes.addAll(currentShard2NodesMap.get(shard)); neededDeployments = chooseNewNodes(currentNode2ShardsMap, orderedNodes, shard, assignedNodes, neededDeployments); if (neededDeployments > 0) { LOG.warn("cannot replicate shard '" + shard + "' " + replicationLevel + " times, cause only " + orderedNodes.size() + " nodes connected"); } else if (neededDeployments < 0) { if (LOG.isDebugEnabled()) { LOG.debug("found shard '" + shard + "' over-replicated"); } removeOverreplicatedShards(currentShard2NodesMap, currentNode2ShardsMap, orderedNodes, 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 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); }