@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); }
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; }
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; } }
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; }  }
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); }
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);  }