You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/10/06 22:58:10 UTC
[flink] 01/02: [FLINK-8532] Modify RebalancePartitioner to use a
random partition as its first partition
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 347482c752e3c693f8a7066d33ee485719e383e0
Author: Guibo Pan <gu...@gmail.com>
AuthorDate: Sun Aug 12 15:11:29 2018 +0800
[FLINK-8532] Modify RebalancePartitioner to use a random partition as its first partition
This closes #6544.
---
.../runtime/partitioner/RebalancePartitioner.java | 15 ++++++++--
.../partitioner/RebalancePartitionerTest.java | 11 ++++---
.../test/streaming/runtime/PartitionerITCase.java | 34 ++++++++++++++++++----
3 files changed, 49 insertions(+), 11 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/partitioner/RebalancePartitioner.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/partitioner/RebalancePartitioner.java
index bb88d17..2c08884 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/partitioner/RebalancePartitioner.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/partitioner/RebalancePartitioner.java
@@ -21,6 +21,8 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import java.util.concurrent.ThreadLocalRandom;
+
/**
* Partitioner that distributes the data equally by cycling through the output
* channels.
@@ -31,18 +33,27 @@ import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
public class RebalancePartitioner<T> extends StreamPartitioner<T> {
private static final long serialVersionUID = 1L;
- private final int[] returnArray = new int[] {-1};
+ private final int[] returnArray = new int[] {Integer.MAX_VALUE - 1};
@Override
public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,
int numberOfOutputChannels) {
int newChannel = ++this.returnArray[0];
if (newChannel >= numberOfOutputChannels) {
- this.returnArray[0] = 0;
+ this.returnArray[0] = resetValue(record, numberOfOutputChannels, newChannel);
}
return this.returnArray;
}
+ private int resetValue(SerializationDelegate<StreamRecord<T>> record,
+ int numberOfOutputChannels, int newChannel) {
+ if (newChannel == Integer.MAX_VALUE) {
+ // Initializes the first partition, this branch is only entered when initializing.
+ return ThreadLocalRandom.current().nextInt(numberOfOutputChannels);
+ }
+ return 0;
+ }
+
public StreamPartitioner<T> copy() {
return this;
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/RebalancePartitionerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/RebalancePartitionerTest.java
index 85410f3..0cf022b 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/RebalancePartitionerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/RebalancePartitionerTest.java
@@ -25,6 +25,7 @@ import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
/**
* Tests for {@link RebalancePartitioner}.
@@ -52,9 +53,11 @@ public class RebalancePartitionerTest {
@Test
public void testSelectChannelsInterval() {
sd.setInstance(streamRecord);
- assertEquals(0, distributePartitioner.selectChannels(sd, 3)[0]);
- assertEquals(1, distributePartitioner.selectChannels(sd, 3)[0]);
- assertEquals(2, distributePartitioner.selectChannels(sd, 3)[0]);
- assertEquals(0, distributePartitioner.selectChannels(sd, 3)[0]);
+ int initialChannel = distributePartitioner.selectChannels(sd, 3)[0];
+ assertTrue(0 <= initialChannel);
+ assertTrue(3 > initialChannel);
+ assertEquals((initialChannel + 1) % 3, distributePartitioner.selectChannels(sd, 3)[0]);
+ assertEquals((initialChannel + 2) % 3, distributePartitioner.selectChannels(sd, 3)[0]);
+ assertEquals((initialChannel + 3) % 3, distributePartitioner.selectChannels(sd, 3)[0]);
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/PartitionerITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/PartitionerITCase.java
index 23fc2eb..cdea26d 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/PartitionerITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/PartitionerITCase.java
@@ -219,18 +219,42 @@ public class PartitionerITCase extends AbstractTestBase {
}
private static void verifyRebalancePartitioning(List<Tuple2<Integer, String>> rebalancePartitionResult) {
- List<Tuple2<Integer, String>> expected = Arrays.asList(
+ List<List<Tuple2<Integer, String>>> expected = Arrays.asList(
+ Arrays.asList(
new Tuple2<Integer, String>(0, "a"),
new Tuple2<Integer, String>(1, "b"),
new Tuple2<Integer, String>(2, "b"),
new Tuple2<Integer, String>(0, "a"),
new Tuple2<Integer, String>(1, "a"),
new Tuple2<Integer, String>(2, "c"),
- new Tuple2<Integer, String>(0, "a"));
+ new Tuple2<Integer, String>(0, "a")),
- assertEquals(
- new HashSet<Tuple2<Integer, String>>(expected),
- new HashSet<Tuple2<Integer, String>>(rebalancePartitionResult));
+ Arrays.asList(
+ new Tuple2<Integer, String>(1, "a"),
+ new Tuple2<Integer, String>(2, "b"),
+ new Tuple2<Integer, String>(0, "b"),
+ new Tuple2<Integer, String>(1, "a"),
+ new Tuple2<Integer, String>(2, "a"),
+ new Tuple2<Integer, String>(0, "c"),
+ new Tuple2<Integer, String>(1, "a")),
+
+ Arrays.asList(
+ new Tuple2<Integer, String>(2, "a"),
+ new Tuple2<Integer, String>(0, "b"),
+ new Tuple2<Integer, String>(1, "b"),
+ new Tuple2<Integer, String>(2, "a"),
+ new Tuple2<Integer, String>(0, "a"),
+ new Tuple2<Integer, String>(1, "c"),
+ new Tuple2<Integer, String>(2, "a")));
+
+ int matchedNum = 0;
+ for (List<Tuple2<Integer, String>> e : expected) {
+ if (new HashSet<Tuple2<Integer, String>>(e).equals(new HashSet<Tuple2<Integer, String>>(rebalancePartitionResult))) {
+ ++matchedNum;
+ }
+ }
+
+ assertEquals(1, matchedNum);
}
private static void verifyGlobalPartitioning(List<Tuple2<Integer, String>> globalPartitionResult) {