public static <T extends Operator>Collection<Partition<T>> repartitionInputOperator(Collection<Partition<T>> partitions){
  List<Partition<T>> newPartitions=new ArrayList<Partition<T>>();
  List<Partition<T>> lowLoadPartitions=new ArrayList<Partition<T>>();
  for (  Partition<T> p : partitions) {
    int load=p.getLoad();
    if (load < 0) {
      if (!lowLoadPartitions.isEmpty()) {
        newPartitions.add(lowLoadPartitions.remove(0));
      }
 else {
        lowLoadPartitions.add(p);
      }
    }
 else     if (load > 0) {
      newPartitions.add(new DefaultPartition<T>(p.getPartitionedInstance()));
      newPartitions.add(new DefaultPartition<T>(p.getPartitionedInstance()));
    }
 else {
      newPartitions.add(p);
    }
  }
  newPartitions.addAll(lowLoadPartitions);
  return newPartitions;
}

public static < T extends Operator > Collection < Partition < T > > repartitionInputOperator ( Collection < Partition < T > > partitions ) { List < Partition < T > > newPartitions = new ArrayList < Partition < T > > ( ) ; List < Partition < T > > lowLoadPartitions = new ArrayList < Partition < T > > ( ) ; for ( Partition < T > p : partitions ) { int load = p . getLoad ( ) ; if ( load < 0 ) { if ( ! lowLoadPartitions . isEmpty ( ) ) { newPartitions . add ( lowLoadPartitions . remove ( 0 ) ) ; } else { lowLoadPartitions . add ( p ) ; } } else if ( load > 0 ) { newPartitions . add ( new DefaultPartition < T > ( p . getPartitionedInstance ( ) ) ) ; newPartitions . add ( new DefaultPartition < T > ( p . getPartitionedInstance ( ) ) ) ; } else { newPartitions . add ( p ) ; } } newPartitions . addAll ( lowLoadPartitions ) ; return newPartitions ; }
adjust the partitions of an input operator ( operator with no connected input stream ) .
