pnowojski commented on a change in pull request #7051: [FLINK-10820][network]
Simplify the RebalancePartitioner implementation
URL: https://github.com/apache/flink/pull/7051#discussion_r232263469
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/partitioner/RebalancePartitioner.java
##########
@@ -33,29 +31,14 @@
public class RebalancePartitioner<T> extends StreamPartitioner<T> {
private static final long serialVersionUID = 1L;
- private final int[] returnArray = {Integer.MAX_VALUE - 1};
+ private final int[] returnArray = new int[] { -1 };
@Override
- public int[] selectChannels(
- SerializationDelegate<StreamRecord<T>> record,
- int numChannels) {
- int newChannel = ++returnArray[0];
- if (newChannel >= numChannels) {
- returnArray[0] = resetValue(numChannels, newChannel);
- }
+ public int[] selectChannels(SerializationDelegate<StreamRecord<T>>
record, int numChannels) {
+ returnArray[0] = (returnArray[0] + 1) % numChannels;
return returnArray;
}
- private static int resetValue(
- int numChannels,
- int newChannel) {
- if (newChannel == Integer.MAX_VALUE) {
- // Initializes the first partition, this branch is only
entered when initializing.
- return ThreadLocalRandom.current().nextInt(numChannels);
Review comment:
Lack of this random changes the actual semantic of the code. With your
change the round robin will always start with the same partition. If you follow
the annotations when it was introduced, it leads to this ticket:
https://issues.apache.org/jira/browse/FLINK-8532
I don't think that this if condition should effect performance in a
meaningful way anyway (but who knows). Nevertheless this random was introduced
for some reason and I would be inclined to keep it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
