Ensure all active nodes are in contention before relinquishing a partition

Change-Id: I846f810547b286736d26d319f315048398334a83

@@ -200,10 +200,14 @@ public class IntentPartitionManager implements IntentPartitionService {
 
 
         for (int i = 0; i < relinquish; i++) {
         for (int i = 0; i < relinquish; i++) {
             String topic = myPartitions.get(i);
             String topic = myPartitions.get(i);
+            // Wait till all active nodes are in contention for partition ownership.
+            // This avoids too many relinquish/reclaim cycles.
+            if (leadershipService.getCandidates(topic).size() == activeNodes) {
                 leadershipService.withdraw(topic);
                 leadershipService.withdraw(topic);
                 executor.schedule(() -> recontest(topic), BACKOFF_TIME, TimeUnit.SECONDS);
                 executor.schedule(() -> recontest(topic), BACKOFF_TIME, TimeUnit.SECONDS);
             }
             }
         }
         }
+    }
 
 
     private void scheduleRebalance(int afterDelaySec) {
     private void scheduleRebalance(int afterDelaySec) {
         if (rebalanceScheduled.compareAndSet(false, true)) {
         if (rebalanceScheduled.compareAndSet(false, true)) {

@@ -116,6 +116,11 @@ public class IntentPartitionManagerTest {
                                                               allNodes))
                                                               allNodes))
                                     .anyTimes();
                                     .anyTimes();
         }
         }
+        for (int i = 0; i < IntentPartitionManager.NUM_PARTITIONS; i++) {
+            expect(leadershipService.getCandidates(ELECTION_PREFIX + i))
+            .andReturn(Arrays.asList(MY_NODE_ID, OTHER_NODE_ID))
+            .anyTimes();
+        }
     }
     }
 
 
     /**
     /**