You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2017/09/25 08:54:24 UTC
flink git commit: [hotfix] Add ProcessOperator tests that verify side
outputs work
Repository: flink
Updated Branches:
refs/heads/master 4afca4b3a -> cb615f1bc
[hotfix] Add ProcessOperator tests that verify side outputs work
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/cb615f1b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/cb615f1b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/cb615f1b
Branch: refs/heads/master
Commit: cb615f1bcf6f5a59543011b5fbdb3c6aa25d2e20
Parents: 4afca4b
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Sun Sep 24 10:26:39 2017 +0200
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Sun Sep 24 15:29:11 2017 +0200
----------------------------------------------------------------------
.../api/operators/KeyedProcessOperatorTest.java | 93 ++++++++++++++++++++
.../api/operators/ProcessOperatorTest.java | 58 ++++++++++++
.../flink/streaming/util/TestHarnessUtil.java | 2 +-
3 files changed, 152 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/cb615f1b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/KeyedProcessOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/KeyedProcessOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/KeyedProcessOperatorTest.java
index 696acfa..8e24c46 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/KeyedProcessOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/KeyedProcessOperatorTest.java
@@ -31,9 +31,12 @@ import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.TestHarnessUtil;
import org.apache.flink.util.Collector;
+import org.apache.flink.util.OutputTag;
import org.apache.flink.util.TestLogger;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.ExpectedException;
import java.util.concurrent.ConcurrentLinkedQueue;
@@ -44,6 +47,9 @@ import static org.junit.Assert.assertEquals;
*/
public class KeyedProcessOperatorTest extends TestLogger {
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
@Test
public void testTimestampAndWatermarkQuerying() throws Exception {
@@ -277,6 +283,93 @@ public class KeyedProcessOperatorTest extends TestLogger {
testHarness.close();
}
+ @Test
+ public void testNullOutputTagRefusal() throws Exception {
+ KeyedProcessOperator<Integer, Integer, String> operator =
+ new KeyedProcessOperator<>(new NullOutputTagEmittingProcessFunction());
+
+ OneInputStreamOperatorTestHarness<Integer, String> testHarness =
+ new KeyedOneInputStreamOperatorTestHarness<>(
+ operator, new IdentityKeySelector<>(), BasicTypeInfo.INT_TYPE_INFO);
+
+ testHarness.setup();
+ testHarness.open();
+
+ testHarness.setProcessingTime(17);
+ try {
+ expectedException.expect(IllegalArgumentException.class);
+ testHarness.processElement(new StreamRecord<>(5));
+ } finally {
+ testHarness.close();
+ }
+ }
+
+ /**
+ * This also verifies that the timestamps ouf side-emitted records is correct.
+ */
+ @Test
+ public void testSideOutput() throws Exception {
+ KeyedProcessOperator<Integer, Integer, String> operator =
+ new KeyedProcessOperator<>(new SideOutputProcessFunction());
+
+ OneInputStreamOperatorTestHarness<Integer, String> testHarness =
+ new KeyedOneInputStreamOperatorTestHarness<>(
+ operator, new IdentityKeySelector<>(), BasicTypeInfo.INT_TYPE_INFO);
+
+ testHarness.setup();
+ testHarness.open();
+
+ testHarness.processElement(new StreamRecord<>(42, 17L /* timestamp */));
+
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
+
+ expectedOutput.add(new StreamRecord<>("IN:42", 17L /* timestamp */));
+
+ TestHarnessUtil.assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
+
+ ConcurrentLinkedQueue<StreamRecord<Integer>> expectedIntSideOutput = new ConcurrentLinkedQueue<>();
+ expectedIntSideOutput.add(new StreamRecord<>(42, 17L /* timestamp */));
+ ConcurrentLinkedQueue<StreamRecord<Integer>> intSideOutput =
+ testHarness.getSideOutput(SideOutputProcessFunction.INTEGER_OUTPUT_TAG);
+ TestHarnessUtil.assertOutputEquals(
+ "Side output was not correct.",
+ expectedIntSideOutput,
+ intSideOutput);
+
+ ConcurrentLinkedQueue<StreamRecord<Long>> expectedLongSideOutput = new ConcurrentLinkedQueue<>();
+ expectedLongSideOutput.add(new StreamRecord<>(42L, 17L /* timestamp */));
+ ConcurrentLinkedQueue<StreamRecord<Long>> longSideOutput =
+ testHarness.getSideOutput(SideOutputProcessFunction.LONG_OUTPUT_TAG);
+ TestHarnessUtil.assertOutputEquals(
+ "Side output was not correct.",
+ expectedLongSideOutput,
+ longSideOutput);
+
+ testHarness.close();
+ }
+
+ private static class NullOutputTagEmittingProcessFunction extends ProcessFunction<Integer, String> {
+
+ @Override
+ public void processElement(Integer value, Context ctx, Collector<String> out) throws Exception {
+ ctx.output(null, value);
+ }
+ }
+
+ private static class SideOutputProcessFunction extends ProcessFunction<Integer, String> {
+
+ static final OutputTag<Integer> INTEGER_OUTPUT_TAG = new OutputTag<Integer>("int-out") {};
+ static final OutputTag<Long> LONG_OUTPUT_TAG = new OutputTag<Long>("long-out") {};
+
+ @Override
+ public void processElement(Integer value, Context ctx, Collector<String> out) throws Exception {
+ out.collect("IN:" + value);
+
+ ctx.output(INTEGER_OUTPUT_TAG, value);
+ ctx.output(LONG_OUTPUT_TAG, value.longValue());
+ }
+ }
+
private static class IdentityKeySelector<T> implements KeySelector<T, T> {
private static final long serialVersionUID = 1L;
http://git-wip-us.apache.org/repos/asf/flink/blob/cb615f1b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/ProcessOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/ProcessOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/ProcessOperatorTest.java
index 5476123..2d9d4d3 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/ProcessOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/ProcessOperatorTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.TestHarnessUtil;
import org.apache.flink.util.Collector;
+import org.apache.flink.util.OutputTag;
import org.apache.flink.util.TestLogger;
import org.junit.Rule;
@@ -117,6 +118,49 @@ public class ProcessOperatorTest extends TestLogger {
}
}
+ /**
+ * This also verifies that the timestamps ouf side-emitted records is correct.
+ */
+ @Test
+ public void testSideOutput() throws Exception {
+ ProcessOperator<Integer, String> operator =
+ new ProcessOperator<>(new SideOutputProcessFunction());
+
+ OneInputStreamOperatorTestHarness<Integer, String> testHarness =
+ new OneInputStreamOperatorTestHarness<>(operator);
+
+ testHarness.setup();
+ testHarness.open();
+
+ testHarness.processElement(new StreamRecord<>(42, 17L /* timestamp */));
+
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
+
+ expectedOutput.add(new StreamRecord<>("IN:42", 17L /* timestamp */));
+
+ TestHarnessUtil.assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
+
+ ConcurrentLinkedQueue<StreamRecord<Integer>> expectedIntSideOutput = new ConcurrentLinkedQueue<>();
+ expectedIntSideOutput.add(new StreamRecord<>(42, 17L /* timestamp */));
+ ConcurrentLinkedQueue<StreamRecord<Integer>> intSideOutput =
+ testHarness.getSideOutput(SideOutputProcessFunction.INTEGER_OUTPUT_TAG);
+ TestHarnessUtil.assertOutputEquals(
+ "Side output was not correct.",
+ expectedIntSideOutput,
+ intSideOutput);
+
+ ConcurrentLinkedQueue<StreamRecord<Long>> expectedLongSideOutput = new ConcurrentLinkedQueue<>();
+ expectedLongSideOutput.add(new StreamRecord<>(42L, 17L /* timestamp */));
+ ConcurrentLinkedQueue<StreamRecord<Long>> longSideOutput =
+ testHarness.getSideOutput(SideOutputProcessFunction.LONG_OUTPUT_TAG);
+ TestHarnessUtil.assertOutputEquals(
+ "Side output was not correct.",
+ expectedLongSideOutput,
+ longSideOutput);
+
+ testHarness.close();
+ }
+
private static class NullOutputTagEmittingProcessFunction extends ProcessFunction<Integer, String> {
@Override
@@ -125,6 +169,20 @@ public class ProcessOperatorTest extends TestLogger {
}
}
+ private static class SideOutputProcessFunction extends ProcessFunction<Integer, String> {
+
+ static final OutputTag<Integer> INTEGER_OUTPUT_TAG = new OutputTag<Integer>("int-out") {};
+ static final OutputTag<Long> LONG_OUTPUT_TAG = new OutputTag<Long>("long-out") {};
+
+ @Override
+ public void processElement(Integer value, Context ctx, Collector<String> out) throws Exception {
+ out.collect("IN:" + value);
+
+ ctx.output(INTEGER_OUTPUT_TAG, value);
+ ctx.output(LONG_OUTPUT_TAG, value.longValue());
+ }
+ }
+
private static class QueryingProcessFunction extends ProcessFunction<Integer, String> {
private static final long serialVersionUID = 1L;
http://git-wip-us.apache.org/repos/asf/flink/blob/cb615f1b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
index 807b68c..a1d732b 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
@@ -57,7 +57,7 @@ public class TestHarnessUtil {
/**
* Compare the two queues containing operator/task output by converting them to an array first.
*/
- public static void assertOutputEquals(String message, Queue<Object> expected, Queue<Object> actual) {
+ public static <T> void assertOutputEquals(String message, Queue<T> expected, Queue<T> actual) {
Assert.assertArrayEquals(message,
expected.toArray(),
actual.toArray());