// Path: kafka-eight-ex/src/main/java/com/ebay/pulsar/druid/firehose/kafka/support/data/RebalanceResult.java
// public class RebalanceResult {
// 	/**
// 	 * indicate how many partitions should be taken totally after each round rebalance for one consumer
// 	 */
// 	private int shouldTaken=-1;
// 	/**
// 	 * all the partitions which can be taken by consumer.
// 	 * coordinator will take partitions from here until one of condition reached: 
// 	 * 1. <code>shouldTaken</code> == <code>getMyPartitions().size()</code>
// 	 * 2. all <code>idlePartitions</code> be taken.
// 	 */
// 	private Set<Integer> idlePartitions=Sets.newHashSet();
// 	/**
// 	 * all the partitions which can be released by consumer.
// 	 * coordinator will fetch partitions from here to release until one of codition reached.
// 	 * 1. <code>shouldTaken</code> == <code>getMyPartitions().size()</code>
// 	 * 2. all <code>shouldRelease</code> be taken. 
// 	 */
// 	private Set<Integer> shouldRelease=Sets.newHashSet();
// 	public int getShouldTaken() {
// 		return shouldTaken;
// 	}
// 	public void setShouldTaken(int shouldTaken) {
// 		this.shouldTaken = shouldTaken;
// 	}
// 	public Set<Integer> getIdlePartitions() {
// 		return idlePartitions;
// 	}
// 	public void setIdlePartitions(Set<Integer> idlePartitions) {
// 		this.idlePartitions = idlePartitions;
// 	}
// 	public Set<Integer> getShouldRelease() {
// 		return shouldRelease;
// 	}
// 	public void setShouldRelease(Set<Integer> shouldRelease) {
// 		this.shouldRelease = shouldRelease;
// 	}
// 	
// 	
// }


import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import com.ebay.pulsar.druid.firehose.kafka.support.data.RebalanceResult;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
/*******************************************************************************
*  Copyright © 2012-2015 eBay Software Foundation
*  This program is licensed under the Apache 2.0 licenses.
*  Please see LICENSE for more information.
*******************************************************************************/
package com.ebay.pulsar.druid.firehose.kafka.support;



/**
 *@author qxing
 * 
 **/
public class DefaultPartitionCoordinator extends BasePartitionCoordinator {
	@Override
	public RebalanceResult calculate(String consumerId, Set<Integer> allPartitions,List<String> allConsumers,Map<String,Set<Integer>> view){
		Set<Integer> allTakenPartitions=Sets.newHashSet(Iterables.concat(view.values()));
		Set<Integer> newPartitions=Sets.difference(allPartitions, allTakenPartitions);
		Collections.sort(allConsumers);
		int thisConsumerIndex=allConsumers.indexOf(consumerId);
		if(thisConsumerIndex<0) throw new RuntimeException("Consumer with id: "+consumerId +" not regist.");
		// calc rebalance count
		int partitionCount = allPartitions.size();
		int minTake = partitionCount / allConsumers.size();
		int left = partitionCount % allConsumers.size();
		int shouldTake = minTake;
		if (thisConsumerIndex < left)
			shouldTake++;
		RebalanceResult rr=new RebalanceResult();
// Path: kafka-eight-ex/src/main/java/com/ebay/pulsar/druid/firehose/kafka/support/data/RebalanceResult.java
// public class RebalanceResult {
// 	/**
// 	 * indicate how many partitions should be taken totally after each round rebalance for one consumer
// 	 */
// 	private int shouldTaken=-1;
// 	/**
// 	 * all the partitions which can be taken by consumer.
// 	 * coordinator will take partitions from here until one of condition reached: 
// 	 * 1. <code>shouldTaken</code> == <code>getMyPartitions().size()</code>
// 	 * 2. all <code>idlePartitions</code> be taken.
// 	 */
// 	private Set<Integer> idlePartitions=Sets.newHashSet();
// 	/**
// 	 * all the partitions which can be released by consumer.
// 	 * coordinator will fetch partitions from here to release until one of codition reached.
// 	 * 1. <code>shouldTaken</code> == <code>getMyPartitions().size()</code>
// 	 * 2. all <code>shouldRelease</code> be taken. 
// 	 */
// 	private Set<Integer> shouldRelease=Sets.newHashSet();
// 	public int getShouldTaken() {
// 		return shouldTaken;
// 	}
// 	public void setShouldTaken(int shouldTaken) {
// 		this.shouldTaken = shouldTaken;
// 	}
// 	public Set<Integer> getIdlePartitions() {
// 		return idlePartitions;
// 	}
// 	public void setIdlePartitions(Set<Integer> idlePartitions) {
// 		this.idlePartitions = idlePartitions;
// 	}
// 	public Set<Integer> getShouldRelease() {
// 		return shouldRelease;
// 	}
// 	public void setShouldRelease(Set<Integer> shouldRelease) {
// 		this.shouldRelease = shouldRelease;
// 	}
// 	
// 	
// }


// Path: kafka-eight-ex/src/main/java/com/ebay/pulsar/druid/firehose/kafka/support/DefaultPartitionCoordinator.java
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import com.ebay.pulsar.druid.firehose.kafka.support.data.RebalanceResult;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;

/*******************************************************************************
*  Copyright © 2012-2015 eBay Software Foundation
*  This program is licensed under the Apache 2.0 licenses.
*  Please see LICENSE for more information.
*******************************************************************************/
package com.ebay.pulsar.druid.firehose.kafka.support;



/**
 *@author qxing
 * 
 **/
public class DefaultPartitionCoordinator extends BasePartitionCoordinator {
	@Override
	public RebalanceResult calculate(String consumerId, Set<Integer> allPartitions,List<String> allConsumers,Map<String,Set<Integer>> view){
		Set<Integer> allTakenPartitions=Sets.newHashSet(Iterables.concat(view.values()));
		Set<Integer> newPartitions=Sets.difference(allPartitions, allTakenPartitions);
		Collections.sort(allConsumers);
		int thisConsumerIndex=allConsumers.indexOf(consumerId);
		if(thisConsumerIndex<0) throw new RuntimeException("Consumer with id: "+consumerId +" not regist.");
		// calc rebalance count
		int partitionCount = allPartitions.size();
		int minTake = partitionCount / allConsumers.size();
		int left = partitionCount % allConsumers.size();
		int shouldTake = minTake;
		if (thisConsumerIndex < left)
			shouldTake++;
		RebalanceResult rr=new RebalanceResult();
		rr.setShouldTaken(shouldTake);