@Test public void testRebalance() { setUpLeadershipService(IntentPartitionManager.NUM_PARTITIONS); leadershipService.withdraw(anyString()); expectLastCall().times(7); replay(leadershipService); partitionManager.activate(); partitionManager.doRebalance(); verify(leadershipService); }
private void rebalance() { int activeNodes = (int) clusterService.getNodes() .stream() .filter(node -> clusterService.getState(node.id()).isActive()) .count(); int myShare = (int) Math.ceil((double) NUM_PARTITIONS / activeNodes); IntStream.range(0, NUM_PARTITIONS) .mapToObj(this::getPartitionPath) .map(leadershipService::getLeadership) .filter(leadership -> !leadership.candidates().contains(localNodeId)) .map(Leadership::topic) .forEach(leadershipService::runForLeadership); List<String> myPartitions = IntStream.range(0, NUM_PARTITIONS) .mapToObj(this::getPartitionPath) .map(leadershipService::getLeadership) .filter(Objects::nonNull) .filter(leadership -> localNodeId.equals(leadership.leaderNodeId())) .map(Leadership::topic) .collect(Collectors.toList()); int relinquish = myPartitions.size() - myShare; for (int i = 0; i < relinquish; i++) { String topic = myPartitions.get(i); if (leadershipService.getCandidates(topic).size() == activeNodes) { leadershipService.withdraw(topic); executor.schedule(() -> recontest(topic), BACKOFF_TIME, TimeUnit.SECONDS); } } }
IntentPartitionManager implements IntentPartitionService { private void rebalance() { int activeNodes = (int) clusterService.getNodes() .stream() .filter(node -> clusterService.getState(node.id()).isActive()) .count(); int myShare = (int) Math.ceil((double) NUM_PARTITIONS / activeNodes); IntStream.range(0, NUM_PARTITIONS) .mapToObj(this::getPartitionPath) .map(leadershipService::getLeadership) .filter(leadership -> !leadership.candidates().contains(localNodeId)) .map(Leadership::topic) .forEach(leadershipService::runForLeadership); List<String> myPartitions = IntStream.range(0, NUM_PARTITIONS) .mapToObj(this::getPartitionPath) .map(leadershipService::getLeadership) .filter(Objects::nonNull) .filter(leadership -> localNodeId.equals(leadership.leaderNodeId())) .map(Leadership::topic) .collect(Collectors.toList()); int relinquish = myPartitions.size() - myShare; for (int i = 0; i < relinquish; i++) { String topic = myPartitions.get(i); if (leadershipService.getCandidates(topic).size() == activeNodes) { leadershipService.withdraw(topic); executor.schedule(() -> recontest(topic), BACKOFF_TIME, TimeUnit.SECONDS); } } } }
IntentPartitionManager implements IntentPartitionService { private void rebalance() { int activeNodes = (int) clusterService.getNodes() .stream() .filter(node -> clusterService.getState(node.id()).isActive()) .count(); int myShare = (int) Math.ceil((double) NUM_PARTITIONS / activeNodes); IntStream.range(0, NUM_PARTITIONS) .mapToObj(this::getPartitionPath) .map(leadershipService::getLeadership) .filter(leadership -> !leadership.candidates().contains(localNodeId)) .map(Leadership::topic) .forEach(leadershipService::runForLeadership); List<String> myPartitions = IntStream.range(0, NUM_PARTITIONS) .mapToObj(this::getPartitionPath) .map(leadershipService::getLeadership) .filter(Objects::nonNull) .filter(leadership -> localNodeId.equals(leadership.leaderNodeId())) .map(Leadership::topic) .collect(Collectors.toList()); int relinquish = myPartitions.size() - myShare; for (int i = 0; i < relinquish; i++) { String topic = myPartitions.get(i); if (leadershipService.getCandidates(topic).size() == activeNodes) { leadershipService.withdraw(topic); executor.schedule(() -> recontest(topic), BACKOFF_TIME, TimeUnit.SECONDS); } } }  }
IntentPartitionManager implements IntentPartitionService { private void rebalance() { int activeNodes = (int) clusterService.getNodes() .stream() .filter(node -> clusterService.getState(node.id()).isActive()) .count(); int myShare = (int) Math.ceil((double) NUM_PARTITIONS / activeNodes); IntStream.range(0, NUM_PARTITIONS) .mapToObj(this::getPartitionPath) .map(leadershipService::getLeadership) .filter(leadership -> !leadership.candidates().contains(localNodeId)) .map(Leadership::topic) .forEach(leadershipService::runForLeadership); List<String> myPartitions = IntStream.range(0, NUM_PARTITIONS) .mapToObj(this::getPartitionPath) .map(leadershipService::getLeadership) .filter(Objects::nonNull) .filter(leadership -> localNodeId.equals(leadership.leaderNodeId())) .map(Leadership::topic) .collect(Collectors.toList()); int relinquish = myPartitions.size() - myShare; for (int i = 0; i < relinquish; i++) { String topic = myPartitions.get(i); if (leadershipService.getCandidates(topic).size() == activeNodes) { leadershipService.withdraw(topic); executor.schedule(() -> recontest(topic), BACKOFF_TIME, TimeUnit.SECONDS); } } }  @Activate void activate(); @Deactivate void deactivate(); @Override boolean isMine(Key intentKey); @Override NodeId getLeader(Key intentKey); @Override void addListener(IntentPartitionEventListener listener); @Override void removeListener(IntentPartitionEventListener listener); }
IntentPartitionManager implements IntentPartitionService { private void rebalance() { int activeNodes = (int) clusterService.getNodes() .stream() .filter(node -> clusterService.getState(node.id()).isActive()) .count(); int myShare = (int) Math.ceil((double) NUM_PARTITIONS / activeNodes); IntStream.range(0, NUM_PARTITIONS) .mapToObj(this::getPartitionPath) .map(leadershipService::getLeadership) .filter(leadership -> !leadership.candidates().contains(localNodeId)) .map(Leadership::topic) .forEach(leadershipService::runForLeadership); List<String> myPartitions = IntStream.range(0, NUM_PARTITIONS) .mapToObj(this::getPartitionPath) .map(leadershipService::getLeadership) .filter(Objects::nonNull) .filter(leadership -> localNodeId.equals(leadership.leaderNodeId())) .map(Leadership::topic) .collect(Collectors.toList()); int relinquish = myPartitions.size() - myShare; for (int i = 0; i < relinquish; i++) { String topic = myPartitions.get(i); if (leadershipService.getCandidates(topic).size() == activeNodes) { leadershipService.withdraw(topic); executor.schedule(() -> recontest(topic), BACKOFF_TIME, TimeUnit.SECONDS); } } }  @Activate void activate(); @Deactivate void deactivate(); @Override boolean isMine(Key intentKey); @Override NodeId getLeader(Key intentKey); @Override void addListener(IntentPartitionEventListener listener); @Override void removeListener(IntentPartitionEventListener listener);  }