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:11 UTC
[flink] 02/02: [hotfix] Cleanup PartitionerITCase
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 5c637a6a071b720b7acf8c725ffcbcb07d2f5745
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Sat Oct 6 20:24:59 2018 +0200
[hotfix] Cleanup PartitionerITCase
---
.../runtime/partitioner/RebalancePartitioner.java | 22 ++--
.../test/streaming/runtime/PartitionerITCase.java | 111 ++++++---------------
2 files changed, 42 insertions(+), 91 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 2c08884..6c5c063 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
@@ -33,23 +33,25 @@ import java.util.concurrent.ThreadLocalRandom;
public class RebalancePartitioner<T> extends StreamPartitioner<T> {
private static final long serialVersionUID = 1L;
- private final int[] returnArray = new int[] {Integer.MAX_VALUE - 1};
+ private final int[] returnArray = {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] = resetValue(record, numberOfOutputChannels, newChannel);
+ public int[] selectChannels(
+ SerializationDelegate<StreamRecord<T>> record,
+ int numChannels) {
+ int newChannel = ++returnArray[0];
+ if (newChannel >= numChannels) {
+ returnArray[0] = resetValue(numChannels, newChannel);
}
- return this.returnArray;
+ return returnArray;
}
- private int resetValue(SerializationDelegate<StreamRecord<T>> record,
- int numberOfOutputChannels, int newChannel) {
+ 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(numberOfOutputChannels);
+ return ThreadLocalRandom.current().nextInt(numChannels);
}
return 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 cdea26d..32f1ece 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
@@ -33,11 +33,16 @@ import org.apache.flink.test.util.AbstractTestBase;
import org.junit.Test;
import java.util.Arrays;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import static java.util.Collections.sort;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
@@ -47,6 +52,10 @@ import static org.junit.Assert.fail;
@SuppressWarnings("serial")
public class PartitionerITCase extends AbstractTestBase {
+ private static final int PARALLELISM = 3;
+
+ private static final List<String> INPUT = Arrays.asList("a", "b", "c", "d", "e", "f", "g");
+
@Test(expected = UnsupportedOperationException.class)
public void testForwardFailsLowToHighParallelism() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -89,17 +98,10 @@ public class PartitionerITCase extends AbstractTestBase {
new TestListResultSink<Tuple2<Integer, String>>();
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(3);
-
- DataStream<Tuple1<String>> src = env.fromElements(
- new Tuple1<String>("a"),
- new Tuple1<String>("b"),
- new Tuple1<String>("b"),
- new Tuple1<String>("a"),
- new Tuple1<String>("a"),
- new Tuple1<String>("c"),
- new Tuple1<String>("a")
- );
+ env.setParallelism(PARALLELISM);
+
+ DataStream<Tuple1<String>> src = env.fromCollection(
+ INPUT.stream().map(Tuple1::of).collect(Collectors.toList()));
// partition by hash
src
@@ -190,86 +192,33 @@ public class PartitionerITCase extends AbstractTestBase {
}
private static void verifyBroadcastPartitioning(List<Tuple2<Integer, String>> broadcastPartitionResult) {
- List<Tuple2<Integer, String>> expected = Arrays.asList(
- new Tuple2<Integer, String>(0, "a"),
- new Tuple2<Integer, String>(0, "b"),
- new Tuple2<Integer, String>(0, "b"),
- new Tuple2<Integer, String>(0, "a"),
- new Tuple2<Integer, String>(0, "a"),
- new Tuple2<Integer, String>(0, "c"),
- new Tuple2<Integer, String>(0, "a"),
- new Tuple2<Integer, String>(1, "a"),
- new Tuple2<Integer, String>(1, "b"),
- new Tuple2<Integer, String>(1, "b"),
- new Tuple2<Integer, String>(1, "a"),
- new Tuple2<Integer, String>(1, "a"),
- new Tuple2<Integer, String>(1, "c"),
- new Tuple2<Integer, String>(1, "a"),
- new Tuple2<Integer, String>(2, "a"),
- new Tuple2<Integer, String>(2, "b"),
- new Tuple2<Integer, String>(2, "b"),
- new Tuple2<Integer, String>(2, "a"),
- new Tuple2<Integer, String>(2, "a"),
- new Tuple2<Integer, String>(2, "c"),
- new Tuple2<Integer, String>(2, "a"));
+ final Set<Tuple2<Integer, String>> expectedResult = INPUT.stream().flatMap(
+ input -> IntStream.range(0, PARALLELISM).mapToObj(
+ i -> Tuple2.of(i, input)))
+ .collect(Collectors.toSet());
assertEquals(
- new HashSet<Tuple2<Integer, String>>(expected),
- new HashSet<Tuple2<Integer, String>>(broadcastPartitionResult));
+ expectedResult,
+ new HashSet<>(broadcastPartitionResult));
}
private static void verifyRebalancePartitioning(List<Tuple2<Integer, String>> rebalancePartitionResult) {
- 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")),
-
- 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;
- }
- }
+ sort(rebalancePartitionResult, Comparator.comparing(o -> o.f1));
+
+ final Tuple2<Integer, String> firstEntry = rebalancePartitionResult.get(0);
+ int offset = firstEntry.f0;
- assertEquals(1, matchedNum);
+ final List<Tuple2<Integer, String>> expected = IntStream.range(0, rebalancePartitionResult.size())
+ .mapToObj(index -> Tuple2.of((offset + index) % PARALLELISM, INPUT.get(index)))
+ .collect(Collectors.toList());
+
+ assertEquals(expected, rebalancePartitionResult);
}
private static void verifyGlobalPartitioning(List<Tuple2<Integer, String>> globalPartitionResult) {
- List<Tuple2<Integer, String>> expected = Arrays.asList(
- new Tuple2<Integer, String>(0, "a"),
- new Tuple2<Integer, String>(0, "b"),
- new Tuple2<Integer, String>(0, "b"),
- new Tuple2<Integer, String>(0, "a"),
- new Tuple2<Integer, String>(0, "a"),
- new Tuple2<Integer, String>(0, "c"),
- new Tuple2<Integer, String>(0, "a"));
+ final List<Tuple2<Integer, String>> expected = INPUT.stream().map(i -> Tuple2.of(0, i)).collect(Collectors.toList());
- assertEquals(
- new HashSet<Tuple2<Integer, String>>(expected),
- new HashSet<Tuple2<Integer, String>>(globalPartitionResult));
+ assertEquals(expected, globalPartitionResult);
}
private static class SubtaskIndexAssigner extends RichMapFunction<Tuple1<String>, Tuple2<Integer, String>> {