question: 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 ; }
target:  Adjust the partitions of an input operator (operator with no connected input stream).