You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2016/03/01 17:22:29 UTC
[1/3] flink git commit: [hotfix] Removed deprecated and commented
code in SourceFunctionTest
Repository: flink
Updated Branches:
refs/heads/master 93733cc00 -> 8d67aa55f
[hotfix] Removed deprecated and commented code in SourceFunctionTest
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/8a50524c
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/8a50524c
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/8a50524c
Branch: refs/heads/master
Commit: 8a50524c791a4024c7fe540c7c0fd911b5e8e8db
Parents: ff80dc0
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Mar 1 16:52:12 2016 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Mar 1 17:02:21 2016 +0100
----------------------------------------------------------------------
.../flink/streaming/api/SourceFunctionTest.java | 21 --------------------
1 file changed, 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/8a50524c/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/SourceFunctionTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/SourceFunctionTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/SourceFunctionTest.java
index 4b99202..946b474 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/SourceFunctionTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/SourceFunctionTest.java
@@ -60,25 +60,4 @@ public class SourceFunctionTest {
7));
assertEquals(expectedList, actualList);
}
-
-// TODO: does not work because we cannot set the internal socket anymore
-// @Test
-// public void socketTextStreamTest() throws Exception {
-// List<String> expectedList = Arrays.asList("a", "b", "c");
-// List<String> actualList = new ArrayList<String>();
-//
-// byte[] data = { 'a', '\n', 'b', '\n', 'c', '\n' };
-//
-// Socket socket = mock(Socket.class);
-// when(socket.getInputStream()).thenReturn(new ByteArrayInputStream(data));
-// when(socket.isClosed()).thenReturn(false);
-// when(socket.isConnected()).thenReturn(true);
-//
-// SocketTextStreamFunction source = new SocketTextStreamFunction("", 0, '\n', 0);
-// source.open(new Configuration());
-// while (!source.reachedEnd()) {
-// actualList.add(source.next());
-// }
-// assertEquals(expectedList, actualList);
-// }
}
[2/3] flink git commit: [FLINK-3554] [streaming] Emit a MAX Watermark
after finite sources finished
Posted by se...@apache.org.
[FLINK-3554] [streaming] Emit a MAX Watermark after finite sources finished
This closes #1750
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ff80dc04
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ff80dc04
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ff80dc04
Branch: refs/heads/master
Commit: ff80dc042aff570b7de2b47697e432e18fce5856
Parents: 93733cc
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Mar 1 14:31:26 2016 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Mar 1 17:02:21 2016 +0100
----------------------------------------------------------------------
.../util/IncrementalLearningSkeletonData.java | 4 +-
.../java/org/apache/flink/cep/CEPITCase.java | 4 +-
.../minicluster/LocalFlinkMiniCluster.scala | 37 +++
.../operators/testutils/DummyEnvironment.java | 3 +-
.../api/operators/StoppableStreamSource.java | 7 +
.../streaming/api/operators/StreamSource.java | 52 +++-
.../streaming/api/watermark/Watermark.java | 15 +-
.../flink/streaming/api/SourceFunctionTest.java | 8 +-
.../operators/StreamSourceOperatorTest.java | 251 +++++++++++++++++++
.../operators/windowing/CoGroupJoinITCase.java | 15 +-
.../operators/windowing/WindowFoldITCase.java | 10 +-
.../streaming/timestamp/TimestampITCase.java | 141 ++++++++++-
.../util/OneInputStreamOperatorTestHarness.java | 2 +-
.../util/TwoInputStreamOperatorTestHarness.java | 3 -
.../streaming/api/scala/CoGroupJoinITCase.scala | 15 +-
.../streaming/api/scala/WindowFoldITCase.scala | 6 +-
16 files changed, 511 insertions(+), 62 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ff80dc04/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/ml/util/IncrementalLearningSkeletonData.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/ml/util/IncrementalLearningSkeletonData.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/ml/util/IncrementalLearningSkeletonData.java
index 1f4bfd8..144af99 100644
--- a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/ml/util/IncrementalLearningSkeletonData.java
+++ b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/ml/util/IncrementalLearningSkeletonData.java
@@ -20,12 +20,12 @@ package org.apache.flink.streaming.examples.ml.util;
public class IncrementalLearningSkeletonData {
public static final String RESULTS = "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" +
- "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "0\n" + "0\n" +
+ "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "0\n" +
"0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" +
"0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" +
"0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" +
"0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" +
- "0\n" + "0\n" + "0\n" + "0\n";
+ "0\n" + "0\n" + "0\n" + "0\n" + "0\n";
private IncrementalLearningSkeletonData() {
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ff80dc04/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java
index 4129e34..40c3c86 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java
@@ -296,9 +296,7 @@ public class CEPITCase extends StreamingMultipleProgramsTestBase {
Tuple2.of(new Event(2, "end", 2.0), 8L),
Tuple2.of(new Event(1, "middle", 5.0), 7L),
Tuple2.of(new Event(3, "middle", 6.0), 9L),
- Tuple2.of(new Event(3, "end", 7.0), 7L),
- // last element for high final watermark
- Tuple2.of(new Event(3, "end", 7.0), 100L)
+ Tuple2.of(new Event(3, "end", 7.0), 7L)
).assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<Tuple2<Event,Long>>() {
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/ff80dc04/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
index c803429..a4c10e7 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
@@ -18,16 +18,23 @@
package org.apache.flink.runtime.minicluster
+import java.util
+
import akka.actor.{ActorRef, ActorSystem}
+import org.apache.flink.api.common.JobID
import org.apache.flink.api.common.io.FileOutputFormat
import org.apache.flink.configuration.{ConfigConstants, Configuration}
import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.io.network.netty.NettyConfig
import org.apache.flink.runtime.jobmanager.{MemoryArchivist, JobManager}
+import org.apache.flink.runtime.messages.JobManagerMessages
+import org.apache.flink.runtime.messages.JobManagerMessages.{StoppingFailure, StoppingResponse, RunningJobsStatus, RunningJobs}
import org.apache.flink.runtime.taskmanager.TaskManager
import org.apache.flink.runtime.util.EnvironmentInformation
+import scala.concurrent.Await
+
/**
* Local Flink mini cluster which executes all [[TaskManager]]s and the [[JobManager]] in the same
* JVM. It extends the [[FlinkMiniCluster]] by having convenience functions to setup Flink's
@@ -211,4 +218,34 @@ class LocalFlinkMiniCluster(
JobManager.ARCHIVE_NAME
}
}
+
+ // --------------------------------------------------------------------------
+ // Actions on running jobs
+ // --------------------------------------------------------------------------
+
+ def currentlyRunningJobs: Iterable[JobID] = {
+ val leader = getLeaderGateway(timeout)
+ val future = leader.ask(JobManagerMessages.RequestRunningJobsStatus, timeout)
+ .mapTo[RunningJobsStatus]
+ Await.result(future, timeout).runningJobs.map(_.getJobId)
+ }
+
+ def getCurrentlyRunningJobsJava(): java.util.List[JobID] = {
+ val list = new java.util.ArrayList[JobID]()
+ currentlyRunningJobs.foreach(list.add)
+ list
+ }
+
+ def stopJob(id: JobID) : Unit = {
+ val leader = getLeaderGateway(timeout)
+ val response = leader.ask(new JobManagerMessages.StopJob(id), timeout)
+ .mapTo[StoppingResponse]
+ val rc = Await.result(response, timeout)
+
+ rc match {
+ case failure: StoppingFailure =>
+ throw new Exception(s"Stopping the job with ID $id failed.", failure.cause)
+ case _ =>
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ff80dc04/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java
index ff6593f..3fcc425 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.operators.testutils;
+import java.util.Collections;
import java.util.Map;
import java.util.concurrent.Future;
@@ -105,7 +106,7 @@ public class DummyEnvironment implements Environment {
@Override
public Map<String, Future<Path>> getDistributedCacheEntries() {
- return null;
+ return Collections.emptyMap();
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/ff80dc04/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StoppableStreamSource.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StoppableStreamSource.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StoppableStreamSource.java
index 927f61f..ce8f6cd 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StoppableStreamSource.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StoppableStreamSource.java
@@ -41,7 +41,14 @@ public class StoppableStreamSource<OUT, SRC extends SourceFunction<OUT> & Stoppa
super(sourceFunction);
}
+ /**
+ * Marks the source a stopped and calls {@link StoppableFunction#stop()} on the user function.
+ */
public void stop() {
+ // important: marking the source as stopped has to happen before the function is stopped.
+ // the flag that tracks this status is volatile, so the memory model also guarantees
+ // the happens-before relationship
+ markCanceledOrStopped();
userFunction.stop();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ff80dc04/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
index 2470417..84f59ed 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
@@ -42,14 +42,19 @@ public class StreamSource<OUT, SRC extends SourceFunction<OUT>>
private transient SourceFunction.SourceContext<OUT> ctx;
+ private transient volatile boolean canceledOrStopped = false;
+
+
public StreamSource(SRC sourceFunction) {
super(sourceFunction);
this.chainingStrategy = ChainingStrategy.HEAD;
}
+
public void run(final Object lockingObject, final Output<StreamRecord<OUT>> collector) throws Exception {
final TimeCharacteristic timeCharacteristic = getOperatorConfig().getTimeCharacteristic();
+ final SourceFunction.SourceContext<OUT> ctx;
switch (timeCharacteristic) {
case EventTime:
@@ -66,19 +71,60 @@ public class StreamSource<OUT, SRC extends SourceFunction<OUT>>
throw new Exception(String.valueOf(timeCharacteristic));
}
- userFunction.run(ctx);
-
- ctx.close();
+ // copy to a field to give the 'cancel()' method access
+ this.ctx = ctx;
+
+ try {
+ userFunction.run(ctx);
+
+ // if we get here, then the user function either exited after being done (finite source)
+ // or the function was canceled or stopped. For the finite source case, we should emit
+ // a final watermark that indicates that we reached the end of event-time
+ if (!isCanceledOrStopped()) {
+ ctx.emitWatermark(Watermark.MAX_WATERMARK);
+ }
+ } finally {
+ // make sure that the context is closed in any case
+ ctx.close();
+ }
}
public void cancel() {
+ // important: marking the source as stopped has to happen before the function is stopped.
+ // the flag that tracks this status is volatile, so the memory model also guarantees
+ // the happens-before relationship
+ markCanceledOrStopped();
userFunction.cancel();
+
// the context may not be initialized if the source was never running.
if (ctx != null) {
ctx.close();
}
}
+
+ /**
+ * Marks this source as canceled or stopped.
+ *
+ * <p>This indicates that any exit of the {@link #run(Object, Output)} method
+ * cannot be interpreted as the result of a finite source.
+ */
+ protected void markCanceledOrStopped() {
+ this.canceledOrStopped = true;
+ }
+ /**
+ * Checks whether the source has been canceled or stopped.
+ * @return True, if the source is canceled or stopped, false is not.
+ */
+ protected boolean isCanceledOrStopped() {
+ return canceledOrStopped;
+ }
+
+ /**
+ * Checks whether any asynchronous thread (checkpoint trigger, timer, watermark generator, ...)
+ * has caused an exception. If one of these threads caused an exception, this method will
+ * throw that exception.
+ */
void checkAsyncException() {
getContainingTask().checkTimerException();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ff80dc04/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/watermark/Watermark.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/watermark/Watermark.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/watermark/Watermark.java
index 1375c0d..cb9eb99 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/watermark/Watermark.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/watermark/Watermark.java
@@ -1,4 +1,4 @@
-/**
+/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
@@ -7,7 +7,7 @@
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.streaming.api.watermark;
import org.apache.flink.annotation.PublicEvolving;
@@ -38,11 +39,15 @@ import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
* <p>
* When a source closes it will emit a final watermark with timestamp {@code Long.MAX_VALUE}. When
* an operator receives this it will know that no more input will be arriving in the future.
- *
*/
@PublicEvolving
-public class Watermark extends StreamElement {
+public final class Watermark extends StreamElement {
+ /** The watermark that signifies end-of-event-time */
+ public static final Watermark MAX_WATERMARK = new Watermark(Long.MAX_VALUE);
+
+ // ------------------------------------------------------------------------
+
/** The timestamp of the watermark */
private final long timestamp;
@@ -60,6 +65,8 @@ public class Watermark extends StreamElement {
return timestamp;
}
+ // ------------------------------------------------------------------------
+
@Override
public boolean equals(Object o) {
return this == o ||
http://git-wip-us.apache.org/repos/asf/flink/blob/ff80dc04/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/SourceFunctionTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/SourceFunctionTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/SourceFunctionTest.java
index b53649a..4b99202 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/SourceFunctionTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/SourceFunctionTest.java
@@ -61,9 +61,9 @@ public class SourceFunctionTest {
assertEquals(expectedList, actualList);
}
- @Test
- public void socketTextStreamTest() throws Exception {
- // TODO: does not work because we cannot set the internal socket anymore
+// TODO: does not work because we cannot set the internal socket anymore
+// @Test
+// public void socketTextStreamTest() throws Exception {
// List<String> expectedList = Arrays.asList("a", "b", "c");
// List<String> actualList = new ArrayList<String>();
//
@@ -80,5 +80,5 @@ public class SourceFunctionTest {
// actualList.add(source.next());
// }
// assertEquals(expectedList, actualList);
- }
+// }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ff80dc04/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamSourceOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamSourceOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamSourceOperatorTest.java
new file mode 100644
index 0000000..b368019
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamSourceOperatorTest.java
@@ -0,0 +1,251 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.operators;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.api.common.functions.StoppableFunction;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.operators.testutils.DummyEnvironment;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.StoppableStreamSource;
+import org.apache.flink.streaming.api.operators.StreamSource;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.Assert.*;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+@SuppressWarnings("serial")
+public class StreamSourceOperatorTest {
+
+ @Test
+ public void testEmitMaxWatermarkForFiniteSource() throws Exception {
+
+ // regular stream source operator
+ StreamSource<String, FiniteSource<String>> operator =
+ new StreamSource<>(new FiniteSource<String>());
+
+ final List<StreamElement> output = new ArrayList<>();
+
+ setupSourceOperator(operator);
+ operator.run(new Object(), new CollectorOutput<String>(output));
+
+ assertEquals(1, output.size());
+ assertEquals(Watermark.MAX_WATERMARK, output.get(0));
+ }
+
+ @Test
+ public void testNoMaxWatermarkOnImmediateCancel() throws Exception {
+
+ final List<StreamElement> output = new ArrayList<>();
+
+ // regular stream source operator
+ final StreamSource<String, InfiniteSource<String>> operator =
+ new StreamSource<>(new InfiniteSource<String>());
+
+
+ setupSourceOperator(operator);
+ operator.cancel();
+
+ // run and exit
+ operator.run(new Object(), new CollectorOutput<String>(output));
+
+ assertTrue(output.isEmpty());
+ }
+
+ @Test
+ public void testNoMaxWatermarkOnAsyncCancel() throws Exception {
+
+ final List<StreamElement> output = new ArrayList<>();
+ final Thread runner = Thread.currentThread();
+
+ // regular stream source operator
+ final StreamSource<String, InfiniteSource<String>> operator =
+ new StreamSource<>(new InfiniteSource<String>());
+
+
+ setupSourceOperator(operator);
+
+ // trigger an async cancel in a bit
+ new Thread("canceler") {
+ @Override
+ public void run() {
+ try {
+ Thread.sleep(200);
+ } catch (InterruptedException ignored) {}
+ operator.cancel();
+ runner.interrupt();
+ }
+ }.start();
+
+ // run and wait to be canceled
+ try {
+ operator.run(new Object(), new CollectorOutput<String>(output));
+ }
+ catch (InterruptedException ignored) {}
+
+ assertTrue(output.isEmpty());
+ }
+
+ @Test
+ public void testNoMaxWatermarkOnImmediateStop() throws Exception {
+
+ final List<StreamElement> output = new ArrayList<>();
+
+ // regular stream source operator
+ final StoppableStreamSource<String, InfiniteSource<String>> operator =
+ new StoppableStreamSource<>(new InfiniteSource<String>());
+
+
+ setupSourceOperator(operator);
+ operator.stop();
+
+ // run and stop
+ operator.run(new Object(), new CollectorOutput<String>(output));
+
+ assertTrue(output.isEmpty());
+ }
+
+ @Test
+ public void testNoMaxWatermarkOnAsyncStop() throws Exception {
+
+ final List<StreamElement> output = new ArrayList<>();
+
+ // regular stream source operator
+ final StoppableStreamSource<String, InfiniteSource<String>> operator =
+ new StoppableStreamSource<>(new InfiniteSource<String>());
+
+
+ setupSourceOperator(operator);
+
+ // trigger an async cancel in a bit
+ new Thread("canceler") {
+ @Override
+ public void run() {
+ try {
+ Thread.sleep(200);
+ } catch (InterruptedException ignored) {}
+ operator.stop();
+ }
+ }.start();
+
+ // run and wait to be stopped
+ operator.run(new Object(), new CollectorOutput<String>(output));
+
+ assertTrue(output.isEmpty());
+ }
+
+
+ // ------------------------------------------------------------------------
+
+ @SuppressWarnings("unchecked")
+ private static <T> void setupSourceOperator(StreamSource<T, ?> operator) {
+ ExecutionConfig executionConfig = new ExecutionConfig();
+ StreamConfig cfg = new StreamConfig(new Configuration());
+
+ cfg.setTimeCharacteristic(TimeCharacteristic.EventTime);
+
+ Environment env = new DummyEnvironment("MockTwoInputTask", 1, 0);
+
+ StreamTask<?, ?> mockTask = mock(StreamTask.class);
+ when(mockTask.getName()).thenReturn("Mock Task");
+ when(mockTask.getCheckpointLock()).thenReturn(new Object());
+ when(mockTask.getConfiguration()).thenReturn(cfg);
+ when(mockTask.getEnvironment()).thenReturn(env);
+ when(mockTask.getExecutionConfig()).thenReturn(executionConfig);
+ when(mockTask.getAccumulatorMap()).thenReturn(Collections.<String, Accumulator<?, ?>>emptyMap());
+
+ operator.setup(mockTask, cfg, (Output< StreamRecord<T>>) mock(Output.class));
+ }
+
+ // ------------------------------------------------------------------------
+
+ private static final class FiniteSource<T> implements SourceFunction<T>, StoppableFunction {
+
+ @Override
+ public void run(SourceContext<T> ctx) {}
+
+ @Override
+ public void cancel() {}
+
+ @Override
+ public void stop() {}
+ }
+
+ private static final class InfiniteSource<T> implements SourceFunction<T>, StoppableFunction {
+
+ private volatile boolean running = true;
+
+ @Override
+ public void run(SourceContext<T> ctx) throws Exception {
+ while (running) {
+ Thread.sleep(20);
+ }
+ }
+
+ @Override
+ public void cancel() {
+ running = false;
+ }
+
+ @Override
+ public void stop() {
+ running = false;
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ private static class CollectorOutput<T> implements Output<StreamRecord<T>> {
+
+ private final List<StreamElement> list;
+
+ private CollectorOutput(List<StreamElement> list) {
+ this.list = list;
+ }
+
+ @Override
+ public void emitWatermark(Watermark mark) {
+ list.add(mark);
+ }
+
+ @Override
+ public void collect(StreamRecord<T> record) {
+ list.add(record);
+ }
+
+ @Override
+ public void close() {}
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/ff80dc04/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/CoGroupJoinITCase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/CoGroupJoinITCase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/CoGroupJoinITCase.java
index c40874c..5e67c72 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/CoGroupJoinITCase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/CoGroupJoinITCase.java
@@ -74,8 +74,7 @@ public class CoGroupJoinITCase extends StreamingMultipleProgramsTestBase {
ctx.collect(Tuple2.of("a", 7));
ctx.collect(Tuple2.of("a", 8));
- // so we get a final big watermark
- ctx.collect(Tuple2.of("a", 20));
+ // source is finite, so it will have an implicit MAX watermark when it finishes
}
@Override
@@ -96,8 +95,7 @@ public class CoGroupJoinITCase extends StreamingMultipleProgramsTestBase {
ctx.collect(Tuple2.of("c", 7));
ctx.collect(Tuple2.of("c", 8));
- // so we get a final big watermark
- ctx.collect(Tuple2.of("a", 20));
+ // source is finite, so it will have an implicit MAX watermark when it finishes
}
@Override
@@ -172,8 +170,7 @@ public class CoGroupJoinITCase extends StreamingMultipleProgramsTestBase {
ctx.collect(Tuple3.of("a", "j", 7));
ctx.collect(Tuple3.of("a", "k", 8));
- // so we get a final big watermark
- ctx.collect(Tuple3.of("a", "k", 20));
+ // source is finite, so it will have an implicit MAX watermark when it finishes
}
@Override
@@ -194,8 +191,7 @@ public class CoGroupJoinITCase extends StreamingMultipleProgramsTestBase {
ctx.collect(Tuple3.of("a", "x", 6));
ctx.collect(Tuple3.of("a", "z", 8));
- // so we get a final high watermark
- ctx.collect(Tuple3.of("a", "z", 20));
+ // source is finite, so it will have an implicit MAX watermark when it finishes
}
@Override
@@ -272,8 +268,7 @@ public class CoGroupJoinITCase extends StreamingMultipleProgramsTestBase {
ctx.collect(Tuple3.of("a", "j", 7));
ctx.collect(Tuple3.of("a", "k", 8));
- // so we get a final high watermark
- ctx.collect(Tuple3.of("a", "k", 20));
+ // source is finite, so it will have an implicit MAX watermark when it finishes
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/ff80dc04/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowFoldITCase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowFoldITCase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowFoldITCase.java
index 3b859f0..fbe03c5 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowFoldITCase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowFoldITCase.java
@@ -74,13 +74,12 @@ public class WindowFoldITCase extends StreamingMultipleProgramsTestBase {
ctx.collect(Tuple2.of("a", 7));
ctx.collect(Tuple2.of("a", 8));
- // so that we get a high final watermark to process the previously sent elements
- ctx.collect(Tuple2.of("a", 20));
+ // source is finite, so it will have an implicit MAX watermark when it finishes
}
@Override
- public void cancel() {
- }
+ public void cancel() {}
+
}).assignTimestampsAndWatermarks(new Tuple2TimestampExtractor());
source1
@@ -139,8 +138,7 @@ public class WindowFoldITCase extends StreamingMultipleProgramsTestBase {
ctx.collect(Tuple2.of("b", 5));
ctx.collect(Tuple2.of("a", 5));
- // so that we get a high final watermark to process the previously sent elements
- ctx.collect(Tuple2.of("a", 20));
+ // source is finite, so it will have an implicit MAX watermark when it finishes
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/ff80dc04/flink-streaming-java/src/test/java/org/apache/flink/streaming/timestamp/TimestampITCase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/timestamp/TimestampITCase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/timestamp/TimestampITCase.java
index d918ba8..1a59ab3 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/timestamp/TimestampITCase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/timestamp/TimestampITCase.java
@@ -18,8 +18,10 @@
package org.apache.flink.streaming.timestamp;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.functions.StoppableFunction;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.ConfigConstants;
@@ -52,7 +54,8 @@ import org.junit.Test;
import java.util.ArrayList;
import java.util.List;
-import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -156,14 +159,88 @@ public class TimestampITCase {
System.err.println(CustomOperator.finalWatermarks[i].get(k));
}
- Assert.fail("Wrong watermark.");
+ fail("Wrong watermark.");
}
}
- assertFalse(CustomOperator.finalWatermarks[i].get(CustomOperator.finalWatermarks[i].size()-1).equals(new Watermark(Long.MAX_VALUE)));
+
+ assertEquals(Watermark.MAX_WATERMARK,
+ CustomOperator.finalWatermarks[i].get(CustomOperator.finalWatermarks[i].size()-1));
}
}
+ @Test
+ public void testWatermarkPropagationNoFinalWatermarkOnStop() throws Exception {
+
+ // for this test to work, we need to be sure that no other jobs are being executed
+ while (!cluster.getCurrentlyRunningJobsJava().isEmpty()) {
+ Thread.sleep(100);
+ }
+
+ final int NUM_WATERMARKS = 10;
+
+ long initialTime = 0L;
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment(
+ "localhost", cluster.getLeaderRPCPort());
+
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+ env.setParallelism(PARALLELISM);
+ env.getConfig().disableSysoutLogging();
+
+ DataStream<Integer> source1 = env.addSource(new MyTimestampSourceInfinite(initialTime, NUM_WATERMARKS));
+ DataStream<Integer> source2 = env.addSource(new MyTimestampSourceInfinite(initialTime, NUM_WATERMARKS / 2));
+
+ source1.union(source2)
+ .map(new IdentityMap())
+ .connect(source2).map(new IdentityCoMap())
+ .transform("Custom Operator", BasicTypeInfo.INT_TYPE_INFO, new CustomOperator(true))
+ .addSink(new NoOpSink<Integer>());
+ new Thread("stopper") {
+ @Override
+ public void run() {
+ try {
+ // try until we get the running jobs
+ List<JobID> running;
+ while ((running = cluster.getCurrentlyRunningJobsJava()).isEmpty()) {
+ Thread.sleep(100);
+ }
+
+ JobID id = running.get(0);
+
+ // send stop until the job is stopped
+ do {
+ cluster.stopJob(id);
+ Thread.sleep(50);
+ } while (!cluster.getCurrentlyRunningJobsJava().isEmpty());
+ }
+ catch (Throwable t) {
+ t.printStackTrace();
+ }
+ }
+ }.start();
+
+ env.execute();
+
+ // verify that all the watermarks arrived at the final custom operator
+ for (int i = 0; i < PARALLELISM; i++) {
+ // we are only guaranteed to see NUM_WATERMARKS / 2 watermarks because the
+ // other source stops emitting after that
+ for (int j = 0; j < NUM_WATERMARKS / 2; j++) {
+ if (!CustomOperator.finalWatermarks[i].get(j).equals(new Watermark(initialTime + j))) {
+ System.err.println("All Watermarks: ");
+ for (int k = 0; k <= NUM_WATERMARKS / 2; k++) {
+ System.err.println(CustomOperator.finalWatermarks[i].get(k));
+ }
+
+ fail("Wrong watermark.");
+ }
+ }
+
+ assertNotEquals(Watermark.MAX_WATERMARK,
+ CustomOperator.finalWatermarks[i].get(CustomOperator.finalWatermarks[i].size()-1));
+ }
+ }
/**
* These check whether timestamps are properly assigned at the sources and handled in
@@ -200,8 +277,7 @@ public class TimestampITCase {
@Test
public void testDisabledTimestamps() throws Exception {
final int NUM_ELEMENTS = 10;
-
-
+
StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment(
"localhost", cluster.getLeaderRPCPort());
@@ -222,7 +298,7 @@ public class TimestampITCase {
}
/**
- * This thests whether timestamps are properly extracted in the timestamp
+ * This tests whether timestamps are properly extracted in the timestamp
* extractor and whether watermarks are also correctly forwared from this with the auto watermark
* interval.
*/
@@ -279,7 +355,10 @@ public class TimestampITCase {
Assert.fail("Wrong watermark. Expected: " + j + " Found: " + wm + " All: " + CustomOperator.finalWatermarks[0]);
}
}
- assertFalse(CustomOperator.finalWatermarks[0].get(CustomOperator.finalWatermarks[0].size()-1).equals(new Watermark(Long.MAX_VALUE)));
+
+ // the input is finite, so it should have a MAX Watermark
+ assertEquals(Watermark.MAX_WATERMARK,
+ CustomOperator.finalWatermarks[0].get(CustomOperator.finalWatermarks[0].size() - 1));
}
/**
@@ -339,7 +418,10 @@ public class TimestampITCase {
Assert.fail("Wrong watermark.");
}
}
- assertFalse(CustomOperator.finalWatermarks[0].get(CustomOperator.finalWatermarks[0].size()-1).equals(new Watermark(Long.MAX_VALUE)));
+
+ // the input is finite, so it should have a MAX Watermark
+ assertEquals(Watermark.MAX_WATERMARK,
+ CustomOperator.finalWatermarks[0].get(CustomOperator.finalWatermarks[0].size() - 1));
}
/**
@@ -400,7 +482,9 @@ public class TimestampITCase {
Assert.fail("Wrong watermark.");
}
}
- assertFalse(CustomOperator.finalWatermarks[0].get(CustomOperator.finalWatermarks[0].size()-1).equals(new Watermark(Long.MAX_VALUE)));
+ // the input is finite, so it should have a MAX Watermark
+ assertEquals(Watermark.MAX_WATERMARK,
+ CustomOperator.finalWatermarks[0].get(CustomOperator.finalWatermarks[0].size() - 1));
}
/**
@@ -710,8 +794,8 @@ public class TimestampITCase {
public static class MyTimestampSource implements SourceFunction<Integer> {
- private long initialTime;
- private int numWatermarks;
+ private final long initialTime;
+ private final int numWatermarks;
public MyTimestampSource(long initialTime, int numWatermarks) {
this.initialTime = initialTime;
@@ -730,6 +814,41 @@ public class TimestampITCase {
public void cancel() {}
}
+ public static class MyTimestampSourceInfinite implements SourceFunction<Integer>, StoppableFunction {
+
+ private final long initialTime;
+ private final int numWatermarks;
+
+ private volatile boolean running = true;
+
+ public MyTimestampSourceInfinite(long initialTime, int numWatermarks) {
+ this.initialTime = initialTime;
+ this.numWatermarks = numWatermarks;
+ }
+
+ @Override
+ public void run(SourceContext<Integer> ctx) throws Exception {
+ for (int i = 0; i < numWatermarks; i++) {
+ ctx.collectWithTimestamp(i, initialTime + i);
+ ctx.emitWatermark(new Watermark(initialTime + i));
+ }
+
+ while (running) {
+ Thread.sleep(20);
+ }
+ }
+
+ @Override
+ public void cancel() {
+ running = false;
+ }
+
+ @Override
+ public void stop() {
+ running = false;
+ }
+ }
+
public static class MyNonWatermarkingSource implements SourceFunction<Integer> {
int numWatermarks;
http://git-wip-us.apache.org/repos/asf/flink/blob/ff80dc04/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OneInputStreamOperatorTestHarness.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OneInputStreamOperatorTestHarness.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OneInputStreamOperatorTestHarness.java
index c484eae..46e74e7 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OneInputStreamOperatorTestHarness.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OneInputStreamOperatorTestHarness.java
@@ -102,7 +102,7 @@ public class OneInputStreamOperatorTestHarness<IN, OUT> {
}
}).when(mockTask).createStateBackend(any(String.class), any(TypeSerializer.class));
} catch (Exception e) {
- e.printStackTrace();
+ throw new RuntimeException(e.getMessage(), e);
}
doAnswer(new Answer<Void>() {
http://git-wip-us.apache.org/repos/asf/flink/blob/ff80dc04/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TwoInputStreamOperatorTestHarness.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TwoInputStreamOperatorTestHarness.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TwoInputStreamOperatorTestHarness.java
index e23673a..d848d2a 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TwoInputStreamOperatorTestHarness.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TwoInputStreamOperatorTestHarness.java
@@ -28,12 +28,9 @@ import org.apache.flink.runtime.operators.testutils.MockInputSplitProvider;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.Output;
import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
-import org.apache.flink.runtime.state.AbstractStateBackend;
-import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.StreamTask;
-import org.mockito.stubbing.OngoingStubbing;
import java.util.concurrent.ConcurrentLinkedQueue;
http://git-wip-us.apache.org/repos/asf/flink/blob/ff80dc04/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/CoGroupJoinITCase.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/CoGroupJoinITCase.scala b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/CoGroupJoinITCase.scala
index f700fa3..fddbe00 100644
--- a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/CoGroupJoinITCase.scala
+++ b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/CoGroupJoinITCase.scala
@@ -56,8 +56,7 @@ class CoGroupJoinITCase extends StreamingMultipleProgramsTestBase {
ctx.collect(("a", 7))
ctx.collect(("a", 8))
- // so that we get a high final watermark to process the previously sent elements
- ctx.collect(("a", 20))
+ // source is finite, so it will have an implicit MAX watermark when it finishes
}
def cancel() {}
@@ -73,8 +72,7 @@ class CoGroupJoinITCase extends StreamingMultipleProgramsTestBase {
ctx.collect(("c", 7))
ctx.collect(("c", 8))
- // so that we get a high final watermark to process the previously sent elements
- ctx.collect(("c", 20))
+ // source is finite, so it will have an implicit MAX watermark when it finishes
}
def cancel() {
@@ -126,8 +124,7 @@ class CoGroupJoinITCase extends StreamingMultipleProgramsTestBase {
ctx.collect(("a", "j", 7))
ctx.collect(("a", "k", 8))
- // so that we get a high final watermark to process the previously sent elements
- ctx.collect(("a", "k", 20))
+ // source is finite, so it will have an implicit MAX watermark when it finishes
}
def cancel() {}
@@ -145,8 +142,7 @@ class CoGroupJoinITCase extends StreamingMultipleProgramsTestBase {
ctx.collect(("a", "x", 6))
ctx.collect(("a", "z", 8))
- // so that we get a high final watermark to process the previously sent elements
- ctx.collect(("a", "z", 20))
+ // source is finite, so it will have an implicit MAX watermark when it finishes
}
def cancel() {}
@@ -208,8 +204,7 @@ class CoGroupJoinITCase extends StreamingMultipleProgramsTestBase {
ctx.collect(("a", "j", 7))
ctx.collect(("a", "k", 8))
- // so that we get a high final watermark to process the previously sent elements
- ctx.collect(("a", "k", 20))
+ // source is finite, so it will have an implicit MAX watermark when it finishes
}
def cancel() {}
http://git-wip-us.apache.org/repos/asf/flink/blob/ff80dc04/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowFoldITCase.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowFoldITCase.scala b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowFoldITCase.scala
index b1d6367..7833651 100644
--- a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowFoldITCase.scala
+++ b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowFoldITCase.scala
@@ -59,8 +59,7 @@ class WindowFoldITCase extends StreamingMultipleProgramsTestBase {
ctx.collect(("a", 7))
ctx.collect(("a", 8))
- // so we get a big watermark to trigger processing of the previous elements
- ctx.collect(("a", 20))
+ // source is finite, so it will have an implicit MAX watermark when it finishes
}
def cancel() {
@@ -107,8 +106,7 @@ class WindowFoldITCase extends StreamingMultipleProgramsTestBase {
ctx.collect(("b", 5))
ctx.collect(("a", 5))
- // so we get a big watermark to trigger processing of the previous elements
- ctx.collect(("a", 20))
+ // source is finite, so it will have an implicit MAX watermark when it finishes
}
def cancel() {
[3/3] flink git commit: [FLINK-2788] [apis] Add TypeHint class to
allow type-safe generic type parsing
Posted by se...@apache.org.
[FLINK-2788] [apis] Add TypeHint class to allow type-safe generic type parsing
This closes #1744
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/8d67aa55
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/8d67aa55
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/8d67aa55
Branch: refs/heads/master
Commit: 8d67aa55f1c11e167a936f1f2a9d954db5960673
Parents: 8a50524
Author: Stephan Ewen <se...@apache.org>
Authored: Mon Feb 29 19:24:34 2016 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Mar 1 17:02:57 2016 +0100
----------------------------------------------------------------------
.../flink/api/common/typeinfo/TypeHint.java | 77 ++++++++
.../api/common/typeinfo/TypeInformation.java | 40 ++++-
.../flink/api/common/typeinfo/TypeHintTest.java | 62 +++++++
.../java/org/apache/flink/api/java/DataSet.java | 1 -
.../java/operators/SingleInputUdfOperator.java | 176 ++++++++++---------
.../api/java/operators/TwoInputUdfOperator.java | 176 +++++++++++--------
.../datastream/SingleOutputStreamOperator.java | 156 ++++++++--------
.../flink/streaming/api/TypeFillTest.java | 48 +++--
8 files changed, 479 insertions(+), 257 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/8d67aa55/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeHint.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeHint.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeHint.java
new file mode 100644
index 0000000..975d6e3
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeHint.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.common.typeinfo;
+
+import org.apache.flink.annotation.Public;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+
+/**
+ * A utility class for describing generic types. It can be used to obtain a type information via:
+ *
+ * <pre>{@code
+ * TypeInformation<Tuple2<String, Long>> info = TypeInformation.of(new TypeHint<Tuple2<String, Long>>(){});
+ * }</pre>
+ * or
+ * <pre>{@code
+ * TypeInformation<Tuple2<String, Long>> info = new TypeHint<Tuple2<String, Long>>(){}.getTypeInfo();
+ * }</pre>
+ *
+ * @param <T> The type information to hint.
+ */
+@Public
+public abstract class TypeHint<T> {
+
+ /** The type information described by the hint */
+ private final TypeInformation<T> typeInfo;
+
+ /**
+ * Creates a hint for the generic type in the class signature.
+ */
+ public TypeHint() {
+ this.typeInfo = TypeExtractor.createTypeInfo(this, TypeHint.class, getClass(), 0);
+ }
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Gets the type information described by this TypeHint.
+ * @return The type information described by this TypeHint.
+ */
+ public TypeInformation<T> getTypeInfo() {
+ return typeInfo;
+ }
+
+ // ------------------------------------------------------------------------
+
+ @Override
+ public int hashCode() {
+ return typeInfo.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj == this ||
+ obj instanceof TypeHint && this.typeInfo.equals(((TypeHint<?>) obj).typeInfo);
+ }
+
+ @Override
+ public String toString() {
+ return "TypeHint: " + typeInfo;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/8d67aa55/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java
index 1c95be0..95eed6b 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java
@@ -22,9 +22,10 @@ import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.annotation.Public;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
import java.io.Serializable;
-import java.util.LinkedList;
+import java.util.Collections;
import java.util.List;
/**
@@ -128,7 +129,7 @@ public abstract class TypeInformation<T> implements Serializable {
@PublicEvolving
public List<TypeInformation<?>> getGenericParameters() {
// Return an empty list as the default implementation
- return new LinkedList<>();
+ return Collections.emptyList();
}
/**
@@ -175,4 +176,39 @@ public abstract class TypeInformation<T> implements Serializable {
* @return true if obj can be equaled with this, otherwise false
*/
public abstract boolean canEqual(Object obj);
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Creates a TypeInformation for the type described by the given class.
+ *
+ * <p>This method only works for non-generic types. For generic types, use the
+ * {@link #of(TypeHint)} method.
+ *
+ * @param typeClass The class of the type.
+ * @param <T> The generic type.
+ *
+ * @return The TypeInformation object for the type described by the hint.
+ */
+ public static <T> TypeInformation<T> of(Class<T> typeClass) {
+ return TypeExtractor.createTypeInfo(typeClass);
+ }
+
+ /**
+ * Creates a TypeInformation for a generic type via a utility "type hint".
+ * This method can be used as follows:
+ * <pre>
+ * {@code
+ * TypeInformation<Tuple2<String, Long>> info = TypeInformation.of(new TypeHint<Tuple2<String, Long>>(){});
+ * }
+ * </pre>
+ *
+ * @param typeHint The hint for the generic type.
+ * @param <T> The generic type.
+ *
+ * @return The TypeInformation object for the type described by the hint.
+ */
+ public static <T> TypeInformation<T> of(TypeHint<T> typeHint) {
+ return typeHint.getTypeInfo();
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8d67aa55/flink-core/src/test/java/org/apache/flink/api/common/typeinfo/TypeHintTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeinfo/TypeHintTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeinfo/TypeHintTest.java
new file mode 100644
index 0000000..60232f2
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeinfo/TypeHintTest.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.common.typeinfo;
+
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+public class TypeHintTest {
+
+ @Test
+ public void testTypeInfoDirect() {
+
+ // simple (non-generic case)
+ TypeHint<String> stringInfo1 = new TypeHint<String>(){};
+ TypeHint<String> stringInfo2 = new TypeHint<String>(){};
+
+ assertEquals(BasicTypeInfo.STRING_TYPE_INFO, stringInfo1.getTypeInfo());
+
+ assertTrue(stringInfo1.hashCode() == stringInfo2.hashCode());
+ assertTrue(stringInfo1.equals(stringInfo2));
+ assertTrue(stringInfo1.toString().equals(stringInfo2.toString()));
+
+ // generic case
+ TypeHint<Tuple3<String, Double, Boolean>> generic = new TypeHint<Tuple3<String, Double, Boolean>>(){};
+
+ TypeInformation<Tuple3<String, Double, Boolean>> tupleInfo =
+ new TupleTypeInfo<>(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.DOUBLE_TYPE_INFO, BasicTypeInfo.BOOLEAN_TYPE_INFO);
+
+ assertEquals(tupleInfo, generic.getTypeInfo());
+ }
+
+ @Test
+ public void testTypeInfoOf() {
+ assertEquals(BasicTypeInfo.STRING_TYPE_INFO, TypeInformation.of(String.class));
+ assertEquals(BasicTypeInfo.STRING_TYPE_INFO, TypeInformation.of(new TypeHint<String>(){}));
+
+
+ TypeInformation<Tuple3<String, Double, Boolean>> tupleInfo =
+ new TupleTypeInfo<>(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.DOUBLE_TYPE_INFO, BasicTypeInfo.BOOLEAN_TYPE_INFO);
+
+ assertEquals(tupleInfo, TypeInformation.of(new TypeHint<Tuple3<String, Double, Boolean>>(){}));
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/8d67aa55/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java b/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java
index c315920..b186c3c 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java
@@ -104,7 +104,6 @@ import java.util.List;
*
* @param <T> The type of the DataSet, i.e., the type of the elements of the DataSet.
*/
-
@Public
public abstract class DataSet<T> {
http://git-wip-us.apache.org/repos/asf/flink/blob/8d67aa55/flink-java/src/main/java/org/apache/flink/api/java/operators/SingleInputUdfOperator.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/SingleInputUdfOperator.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/SingleInputUdfOperator.java
index 22cf089..eb485fe 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/operators/SingleInputUdfOperator.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/SingleInputUdfOperator.java
@@ -26,18 +26,21 @@ import java.util.Set;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.Public;
+import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.functions.Function;
import org.apache.flink.api.common.functions.InvalidTypesException;
import org.apache.flink.api.common.operators.SemanticProperties;
import org.apache.flink.api.common.operators.SingleInputSemanticProperties;
+import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.functions.FunctionAnnotation;
import org.apache.flink.api.java.functions.SemanticPropUtil;
-import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.api.java.typeutils.TypeInfoParser;
import org.apache.flink.configuration.Configuration;
+import static java.util.Objects.requireNonNull;
+
/**
* The <tt>SingleInputUdfOperator</tt> is the base class of all unary operators that execute
* user-defined functions (UDFs). The UDFs encapsulated by this operator are naturally UDFs that
@@ -185,6 +188,92 @@ public abstract class SingleInputUdfOperator<IN, OUT, O extends SingleInputUdfOp
O returnType = (O) this;
return returnType;
}
+
+ // ------------------------------------------------------------------------
+ // type hinting
+ // ------------------------------------------------------------------------
+
+ /**
+ * Adds a type information hint about the return type of this operator. This method
+ * can be used in cases where Flink cannot determine automatically what the produced
+ * type of a function is. That can be the case if the function uses generic type variables
+ * in the return type that cannot be inferred from the input type.
+ *
+ * <p>Classes can be used as type hints for non-generic types (classes without generic parameters),
+ * but not for generic types like for example Tuples. For those generic types, please
+ * use the {@link #returns(TypeHint)} method.
+ *
+ * <p>Use this method the following way:
+ * <pre>{@code
+ * DataSet<String[]> result =
+ * data.flatMap(new FunctionWithNonInferrableReturnType())
+ * .returns(String[].class);
+ * }</pre>
+ *
+ * @param typeClass The class of the returned data type.
+ * @return This operator with the type information corresponding to the given type class.
+ */
+ public O returns(Class<OUT> typeClass) {
+ requireNonNull(typeClass, "type class must not be null");
+
+ try {
+ return returns(TypeInformation.of(typeClass));
+ }
+ catch (InvalidTypesException e) {
+ throw new InvalidTypesException("Cannot infer the type information from the class alone." +
+ "This is most likely because the class represents a generic type. In that case," +
+ "please use the 'returns(TypeHint)' method instead.", e);
+ }
+ }
+
+ /**
+ * Adds a type information hint about the return type of this operator. This method
+ * can be used in cases where Flink cannot determine automatically what the produced
+ * type of a function is. That can be the case if the function uses generic type variables
+ * in the return type that cannot be inferred from the input type.
+ *
+ * <p>Use this method the following way:
+ * <pre>{@code
+ * DataSet<Tuple2<String, Double>> result =
+ * data.flatMap(new FunctionWithNonInferrableReturnType())
+ * .returns(new TypeHint<Tuple2<String, Double>>(){});
+ * }</pre>
+ *
+ * @param typeHint The type hint for the returned data type.
+ * @return This operator with the type information corresponding to the given type hint.
+ */
+ public O returns(TypeHint<OUT> typeHint) {
+ requireNonNull(typeHint, "TypeHint must not be null");
+
+ try {
+ return returns(TypeInformation.of(typeHint));
+ }
+ catch (InvalidTypesException e) {
+ throw new InvalidTypesException("Cannot infer the type information from the type hint. " +
+ "Make sure that the TypeHint does not use any generic type variables.");
+ }
+ }
+
+ /**
+ * Adds a type information hint about the return type of this operator. This method
+ * can be used in cases where Flink cannot determine automatically what the produced
+ * type of a function is. That can be the case if the function uses generic type variables
+ * in the return type that cannot be inferred from the input type.
+ *
+ * <p>In most cases, the methods {@link #returns(Class)} and {@link #returns(TypeHint)}
+ * are preferable.
+ *
+ * @param typeInfo The type information for the returned data type.
+ * @return This operator using the given type information for the return type.
+ */
+ public O returns(TypeInformation<OUT> typeInfo) {
+ requireNonNull(typeInfo, "TypeInformation must not be null");
+
+ fillInType(typeInfo);
+ @SuppressWarnings("unchecked")
+ O returnType = (O) this;
+ return returnType;
+ }
/**
* Adds a type information hint about the return type of this operator.
@@ -220,7 +309,11 @@ public abstract class SingleInputUdfOperator<IN, OUT, O extends SingleInputUdfOp
* @param typeInfoString
* type information string to be parsed
* @return This operator with a given return type hint.
+ *
+ * @deprecated Please use {@link #returns(Class)} or {@link #returns(TypeHint)} instead.
*/
+ @Deprecated
+ @PublicEvolving
public O returns(String typeInfoString) {
if (typeInfoString == null) {
throw new IllegalArgumentException("Type information string must not be null.");
@@ -228,78 +321,6 @@ public abstract class SingleInputUdfOperator<IN, OUT, O extends SingleInputUdfOp
return returns(TypeInfoParser.<OUT>parse(typeInfoString));
}
- /**
- * Adds a type information hint about the return type of this operator.
- *
- * <p>
- * Type hints are important in cases where the Java compiler
- * throws away generic type information necessary for efficient execution.
- *
- * <p>
- * This method takes an instance of {@link org.apache.flink.api.common.typeinfo.TypeInformation} such as:
- *
- * <ul>
- * <li>{@link org.apache.flink.api.common.typeinfo.BasicTypeInfo}</li>
- * <li>{@link org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo}</li>
- * <li>{@link org.apache.flink.api.java.typeutils.TupleTypeInfo}</li>
- * <li>{@link org.apache.flink.api.java.typeutils.PojoTypeInfo}</li>
- * <li>{@link org.apache.flink.api.java.typeutils.WritableTypeInfo}</li>
- * <li>{@link org.apache.flink.api.java.typeutils.ValueTypeInfo}</li>
- * <li>etc.</li>
- * </ul>
- *
- * @param typeInfo
- * type information as a return type hint
- * @return This operator with a given return type hint.
- */
- public O returns(TypeInformation<OUT> typeInfo) {
- if (typeInfo == null) {
- throw new IllegalArgumentException("Type information must not be null.");
- }
- fillInType(typeInfo);
- @SuppressWarnings("unchecked")
- O returnType = (O) this;
- return returnType;
- }
-
- /**
- * Adds a type information hint about the return type of this operator.
- *
- * <p>
- * Type hints are important in cases where the Java compiler
- * throws away generic type information necessary for efficient execution.
- *
- * <p>
- * This method takes a class that will be analyzed by Flink's type extraction capabilities.
- *
- * <p>
- * Examples for classes are:
- * <ul>
- * <li>Basic types such as <code>Integer.class</code>, <code>String.class</code>, etc.</li>
- * <li>POJOs such as <code>MyPojo.class</code></li>
- * <li>Classes that <b>extend</b> tuples. Classes like <code>Tuple1.class</code>,<code>Tuple2.class</code>, etc. are <b>not</b> sufficient.</li>
- * <li>Arrays such as <code>String[].class</code>, etc.</li>
- * </ul>
- *
- * @param typeClass
- * class as a return type hint
- * @return This operator with a given return type hint.
- */
- @SuppressWarnings("unchecked")
- public O returns(Class<OUT> typeClass) {
- if (typeClass == null) {
- throw new IllegalArgumentException("Type class must not be null.");
- }
-
- try {
- TypeInformation<OUT> ti = (TypeInformation<OUT>) TypeExtractor.createTypeInfo(typeClass);
- return returns(ti);
- }
- catch (InvalidTypesException e) {
- throw new InvalidTypesException("The given class is not suited for providing necessary type information.", e);
- }
- }
-
// --------------------------------------------------------------------------------------------
// Accessors
// --------------------------------------------------------------------------------------------
@@ -360,12 +381,7 @@ public abstract class SingleInputUdfOperator<IN, OUT, O extends SingleInputUdfOp
}
protected boolean udfWithForwardedFieldsAnnotation(Class<?> udfClass) {
-
- if (udfClass.getAnnotation(FunctionAnnotation.ForwardedFields.class) != null ||
- udfClass.getAnnotation(FunctionAnnotation.NonForwardedFields.class) != null) {
- return true;
- } else {
- return false;
- }
+ return udfClass.getAnnotation(FunctionAnnotation.ForwardedFields.class) != null ||
+ udfClass.getAnnotation(FunctionAnnotation.NonForwardedFields.class) != null;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8d67aa55/flink-java/src/main/java/org/apache/flink/api/java/operators/TwoInputUdfOperator.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/TwoInputUdfOperator.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/TwoInputUdfOperator.java
index 34a5518..695ed3a 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/operators/TwoInputUdfOperator.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/TwoInputUdfOperator.java
@@ -26,18 +26,21 @@ import java.util.Set;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.Public;
+import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.functions.Function;
import org.apache.flink.api.common.functions.InvalidTypesException;
import org.apache.flink.api.common.operators.DualInputSemanticProperties;
import org.apache.flink.api.common.operators.SemanticProperties;
+import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.functions.FunctionAnnotation;
import org.apache.flink.api.java.functions.SemanticPropUtil;
-import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.api.java.typeutils.TypeInfoParser;
import org.apache.flink.configuration.Configuration;
+import static java.util.Objects.requireNonNull;
+
/**
* The <tt>TwoInputUdfOperator</tt> is the base class of all binary operators that execute
* user-defined functions (UDFs). The UDFs encapsulated by this operator are naturally UDFs that
@@ -262,14 +265,102 @@ public abstract class TwoInputUdfOperator<IN1, IN2, OUT, O extends TwoInputUdfOp
O returnType = (O) this;
return returnType;
}
-
+
+ // ------------------------------------------------------------------------
+ // type hinting
+ // ------------------------------------------------------------------------
+
+ /**
+ * Adds a type information hint about the return type of this operator. This method
+ * can be used in cases where Flink cannot determine automatically what the produced
+ * type of a function is. That can be the case if the function uses generic type variables
+ * in the return type that cannot be inferred from the input type.
+ *
+ * <p>Classes can be used as type hints for non-generic types (classes without generic parameters),
+ * but not for generic types like for example Tuples. For those generic types, please
+ * use the {@link #returns(TypeHint)} method.
+ *
+ * <p>Use this method the following way:
+ * <pre>{@code
+ * DataSet<String[]> result =
+ * data1.join(data2).where("id").equalTo("fieldX")
+ * .with(new JoinFunctionWithNonInferrableReturnType())
+ * .returns(String[].class);
+ * }</pre>
+ *
+ * @param typeClass The class of the returned data type.
+ * @return This operator with the type information corresponding to the given type class.
+ */
+ public O returns(Class<OUT> typeClass) {
+ requireNonNull(typeClass, "type class must not be null");
+
+ try {
+ return returns(TypeInformation.of(typeClass));
+ }
+ catch (InvalidTypesException e) {
+ throw new InvalidTypesException("Cannot infer the type information from the class alone." +
+ "This is most likely because the class represents a generic type. In that case," +
+ "please use the 'returns(TypeHint)' method instead.", e);
+ }
+ }
+
+ /**
+ * Adds a type information hint about the return type of this operator. This method
+ * can be used in cases where Flink cannot determine automatically what the produced
+ * type of a function is. That can be the case if the function uses generic type variables
+ * in the return type that cannot be inferred from the input type.
+ *
+ * <p>Use this method the following way:
+ * <pre>{@code
+ * DataSet<Tuple2<String, Double>> result =
+ * data1.join(data2).where("id").equalTo("fieldX")
+ * .with(new JoinFunctionWithNonInferrableReturnType())
+ * .returns(new TypeHint<Tuple2<String, Double>>(){});
+ * }</pre>
+ *
+ * @param typeHint The type hint for the returned data type.
+ * @return This operator with the type information corresponding to the given type hint.
+ */
+ public O returns(TypeHint<OUT> typeHint) {
+ requireNonNull(typeHint, "TypeHint must not be null");
+
+ try {
+ return returns(TypeInformation.of(typeHint));
+ }
+ catch (InvalidTypesException e) {
+ throw new InvalidTypesException("Cannot infer the type information from the type hint. " +
+ "Make sure that the TypeHint does not use any generic type variables.");
+ }
+ }
+
+ /**
+ * Adds a type information hint about the return type of this operator. This method
+ * can be used in cases where Flink cannot determine automatically what the produced
+ * type of a function is. That can be the case if the function uses generic type variables
+ * in the return type that cannot be inferred from the input type.
+ *
+ * <p>In most cases, the methods {@link #returns(Class)} and {@link #returns(TypeHint)}
+ * are preferable.
+ *
+ * @param typeInfo The type information for the returned data type.
+ * @return This operator using the given type information for the return type.
+ */
+ public O returns(TypeInformation<OUT> typeInfo) {
+ requireNonNull(typeInfo, "TypeInformation must not be null");
+
+ fillInType(typeInfo);
+ @SuppressWarnings("unchecked")
+ O returnType = (O) this;
+ return returnType;
+ }
+
/**
* Adds a type information hint about the return type of this operator.
- *
+ *
* <p>
* Type hints are important in cases where the Java compiler
* throws away generic type information necessary for efficient execution.
- *
+ *
* <p>
* This method takes a type information string that will be parsed. A type information string can contain the following
* types:
@@ -297,86 +388,17 @@ public abstract class TwoInputUdfOperator<IN1, IN2, OUT, O extends TwoInputUdfOp
* @param typeInfoString
* type information string to be parsed
* @return This operator with a given return type hint.
+ *
+ * @deprecated Please use {@link #returns(Class)} or {@link #returns(TypeHint)} instead.
*/
+ @Deprecated
+ @PublicEvolving
public O returns(String typeInfoString) {
if (typeInfoString == null) {
throw new IllegalArgumentException("Type information string must not be null.");
}
return returns(TypeInfoParser.<OUT>parse(typeInfoString));
}
-
- /**
- * Adds a type information hint about the return type of this operator.
- *
- * <p>
- * Type hints are important in cases where the Java compiler
- * throws away generic type information necessary for efficient execution.
- *
- * <p>
- * This method takes an instance of {@link org.apache.flink.api.common.typeinfo.TypeInformation} such as:
- *
- * <ul>
- * <li>{@link org.apache.flink.api.common.typeinfo.BasicTypeInfo}</li>
- * <li>{@link org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo}</li>
- * <li>{@link org.apache.flink.api.java.typeutils.TupleTypeInfo}</li>
- * <li>{@link org.apache.flink.api.java.typeutils.PojoTypeInfo}</li>
- * <li>{@link org.apache.flink.api.java.typeutils.WritableTypeInfo}</li>
- * <li>{@link org.apache.flink.api.java.typeutils.ValueTypeInfo}</li>
- * <li>etc.</li>
- * </ul>
- *
- * @param typeInfo
- * type information as a return type hint
- * @return This operator with a given return type hint.
- */
- public O returns(TypeInformation<OUT> typeInfo) {
- if (typeInfo == null) {
- throw new IllegalArgumentException("Type information must not be null.");
- }
- fillInType(typeInfo);
-
- @SuppressWarnings("unchecked")
- O returnType = (O) this;
- return returnType;
- }
-
- /**
- * Adds a type information hint about the return type of this operator.
- *
- * <p>
- * Type hints are important in cases where the Java compiler
- * throws away generic type information necessary for efficient execution.
- *
- * <p>
- * This method takes a class that will be analyzed by Flink's type extraction capabilities.
- *
- * <p>
- * Examples for classes are:
- * <ul>
- * <li>Basic types such as <code>Integer.class</code>, <code>String.class</code>, etc.</li>
- * <li>POJOs such as <code>MyPojo.class</code></li>
- * <li>Classes that <b>extend</b> tuples. Classes like <code>Tuple1.class</code>,<code>Tuple2.class</code>, etc. are <b>not</b> sufficient.</li>
- * <li>Arrays such as <code>String[].class</code>, etc.</li>
- * </ul>
- *
- * @param typeClass
- * class as a return type hint
- * @return This operator with a given return type hint.
- */
- @SuppressWarnings("unchecked")
- public O returns(Class<OUT> typeClass) {
- if (typeClass == null) {
- throw new IllegalArgumentException("Type class must not be null.");
- }
-
- try {
- TypeInformation<OUT> ti = (TypeInformation<OUT>) TypeExtractor.createTypeInfo(typeClass);
- return returns(ti);
- }
- catch (InvalidTypesException e) {
- throw new InvalidTypesException("The given class is not suited for providing necessary type information.", e);
- }
- }
// --------------------------------------------------------------------------------------------
// Accessors
http://git-wip-us.apache.org/repos/asf/flink/blob/8d67aa55/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java
index 5846c26..2c7b5cb 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java
@@ -20,8 +20,8 @@ package org.apache.flink.streaming.api.datastream;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.annotation.Public;
import org.apache.flink.api.common.functions.InvalidTypesException;
+import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.api.java.typeutils.TypeInfoParser;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.operators.ChainingStrategy;
@@ -29,6 +29,8 @@ import org.apache.flink.streaming.api.transformations.PartitionTransformation;
import org.apache.flink.streaming.api.transformations.StreamTransformation;
import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner;
+import static java.util.Objects.requireNonNull;
+
/**
* {@code SingleOutputStreamOperator} represents a user defined transformation
* applied on a {@link DataStream} with one predefined output type.
@@ -152,6 +154,83 @@ public class SingleOutputStreamOperator<T> extends DataStream<T> {
return setChainingStrategy(ChainingStrategy.HEAD);
}
+ // ------------------------------------------------------------------------
+ // Type hinting
+ // ------------------------------------------------------------------------
+
+ /**
+ * Adds a type information hint about the return type of this operator. This method
+ * can be used in cases where Flink cannot determine automatically what the produced
+ * type of a function is. That can be the case if the function uses generic type variables
+ * in the return type that cannot be inferred from the input type.
+ *
+ * <p>Classes can be used as type hints for non-generic types (classes without generic parameters),
+ * but not for generic types like for example Tuples. For those generic types, please
+ * use the {@link #returns(TypeHint)} method.
+ *
+ * @param typeClass The class of the returned data type.
+ * @return This operator with the type information corresponding to the given type class.
+ */
+ public SingleOutputStreamOperator<T> returns(Class<T> typeClass) {
+ requireNonNull(typeClass, "type class must not be null.");
+
+ try {
+ return returns(TypeInformation.of(typeClass));
+ }
+ catch (InvalidTypesException e) {
+ throw new InvalidTypesException("Cannot infer the type information from the class alone." +
+ "This is most likely because the class represents a generic type. In that case," +
+ "please use the 'returns(TypeHint)' method instead.");
+ }
+ }
+
+ /**
+ * Adds a type information hint about the return type of this operator. This method
+ * can be used in cases where Flink cannot determine automatically what the produced
+ * type of a function is. That can be the case if the function uses generic type variables
+ * in the return type that cannot be inferred from the input type.
+ *
+ * <p>Use this method the following way:
+ * <pre>{@code
+ * DataStream<Tuple2<String, Double>> result =
+ * stream.flatMap(new FunctionWithNonInferrableReturnType())
+ * .returns(new TypeHint<Tuple2<String, Double>>(){});
+ * }</pre>
+ *
+ * @param typeHint The type hint for the returned data type.
+ * @return This operator with the type information corresponding to the given type hint.
+ */
+ public SingleOutputStreamOperator<T> returns(TypeHint<T> typeHint) {
+ requireNonNull(typeHint, "TypeHint must not be null");
+
+ try {
+ return returns(TypeInformation.of(typeHint));
+ }
+ catch (InvalidTypesException e) {
+ throw new InvalidTypesException("Cannot infer the type information from the type hint. " +
+ "Make sure that the TypeHint does not use any generic type variables.");
+ }
+ }
+
+ /**
+ * Adds a type information hint about the return type of this operator. This method
+ * can be used in cases where Flink cannot determine automatically what the produced
+ * type of a function is. That can be the case if the function uses generic type variables
+ * in the return type that cannot be inferred from the input type.
+ *
+ * <p>In most cases, the methods {@link #returns(Class)} and {@link #returns(TypeHint)}
+ * are preferable.
+ *
+ * @param typeInfo type information as a return type hint
+ * @return This operator with a given return type hint.
+ */
+ public SingleOutputStreamOperator<T> returns(TypeInformation<T> typeInfo) {
+ requireNonNull(typeInfo, "TypeInformation must not be null");
+
+ transformation.setOutputType(typeInfo);
+ return this;
+ }
+
/**
* Adds a type information hint about the return type of this operator.
*
@@ -186,7 +265,11 @@ public class SingleOutputStreamOperator<T> extends DataStream<T> {
* @param typeInfoString
* type information string to be parsed
* @return This operator with a given return type hint.
+ *
+ * @deprecated Please use {@link #returns(Class)} or {@link #returns(TypeHint)} instead.
*/
+ @Deprecated
+ @PublicEvolving
public SingleOutputStreamOperator<T> returns(String typeInfoString) {
if (typeInfoString == null) {
throw new IllegalArgumentException("Type information string must not be null.");
@@ -194,74 +277,9 @@ public class SingleOutputStreamOperator<T> extends DataStream<T> {
return returns(TypeInfoParser.<T>parse(typeInfoString));
}
- /**
- * Adds a type information hint about the return type of this operator.
- *
- * <p>
- * Type hints are important in cases where the Java compiler
- * throws away generic type information necessary for efficient execution.
- *
- * <p>
- * This method takes an instance of {@link org.apache.flink.api.common.typeinfo.TypeInformation} such as:
- *
- * <ul>
- * <li>{@link org.apache.flink.api.common.typeinfo.BasicTypeInfo}</li>
- * <li>{@link org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo}</li>
- * <li>{@link org.apache.flink.api.java.typeutils.TupleTypeInfo}</li>
- * <li>{@link org.apache.flink.api.java.typeutils.PojoTypeInfo}</li>
- * <li>{@link org.apache.flink.api.java.typeutils.WritableTypeInfo}</li>
- * <li>{@link org.apache.flink.api.java.typeutils.ValueTypeInfo}</li>
- * <li>etc.</li>
- * </ul>
- *
- * @param typeInfo type information as a return type hint
- * @return This operator with a given return type hint.
- */
- public SingleOutputStreamOperator<T> returns(TypeInformation<T> typeInfo) {
- if (typeInfo == null) {
- throw new IllegalArgumentException("Type information must not be null.");
- }
- transformation.setOutputType(typeInfo);
- return this;
- }
-
- /**
- * Adds a type information hint about the return type of this operator.
- *
- * <p>
- * Type hints are important in cases where the Java compiler
- * throws away generic type information necessary for efficient execution.
- *
- * <p>
- * This method takes a class that will be analyzed by Flink's type extraction capabilities.
- *
- * <p>
- * Examples for classes are:
- * <ul>
- * <li>Basic types such as <code>Integer.class</code>, <code>String.class</code>, etc.</li>
- * <li>POJOs such as <code>MyPojo.class</code></li>
- * <li>Classes that <b>extend</b> tuples. Classes like <code>Tuple1.class</code>,<code>Tuple2.class</code>, etc. are <b>not</b> sufficient.</li>
- * <li>Arrays such as <code>String[].class</code>, etc.</li>
- * </ul>
- *
- * @param typeClass
- * class as a return type hint
- * @return This operator with a given return type hint.
- */
- @SuppressWarnings("unchecked")
- public SingleOutputStreamOperator<T> returns(Class<T> typeClass) {
- if (typeClass == null) {
- throw new IllegalArgumentException("Type class must not be null.");
- }
-
- try {
- TypeInformation<T> ti = (TypeInformation<T>) TypeExtractor.createTypeInfo(typeClass);
- return returns(ti);
- }
- catch (InvalidTypesException e) {
- throw new InvalidTypesException("The given class is not suited for providing necessary type information.", e);
- }
- }
+ // ------------------------------------------------------------------------
+ // Miscellaneous
+ // ------------------------------------------------------------------------
@Override
protected DataStream<T> setConnectionType(StreamPartitioner<T> partitioner) {
http://git-wip-us.apache.org/repos/asf/flink/blob/8d67aa55/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/TypeFillTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/TypeFillTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/TypeFillTest.java
index ddda82d..acbb5b4 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/TypeFillTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/TypeFillTest.java
@@ -24,6 +24,7 @@ import static org.junit.Assert.fail;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -44,36 +45,35 @@ public class TypeFillTest extends StreamingMultipleProgramsTestBase {
try {
env.addSource(new TestSource<Integer>()).print();
fail();
- } catch (Exception e) {
- }
+ } catch (Exception ignored) {}
+
DataStream<Long> source = env.generateSequence(1, 10);
try {
source.map(new TestMap<Long, Long>()).print();
fail();
- } catch (Exception e) {
- }
+ } catch (Exception ignored) {}
+
try {
source.flatMap(new TestFlatMap<Long, Long>()).print();
fail();
- } catch (Exception e) {
- }
+ } catch (Exception ignored) {}
+
try {
source.connect(source).map(new TestCoMap<Long, Long, Integer>()).print();
fail();
- } catch (Exception e) {
- }
+ } catch (Exception ignored) {}
+
try {
source.connect(source).flatMap(new TestCoFlatMap<Long, Long, Integer>()).print();
fail();
- } catch (Exception e) {
- }
+ } catch (Exception ignored) {}
- env.addSource(new TestSource<Integer>()).returns("Integer");
+ env.addSource(new TestSource<Integer>()).returns(Integer.class);
source.map(new TestMap<Long, Long>()).returns(Long.class).print();
- source.flatMap(new TestFlatMap<Long, Long>()).returns("Long").print();
- source.connect(source).map(new TestCoMap<Long, Long, Integer>()).returns("Integer").print();
+ source.flatMap(new TestFlatMap<Long, Long>()).returns(new TypeHint<Long>(){}).print();
+ source.connect(source).map(new TestCoMap<Long, Long, Integer>()).returns(BasicTypeInfo.INT_TYPE_INFO).print();
source.connect(source).flatMap(new TestCoFlatMap<Long, Long, Integer>())
.returns(BasicTypeInfo.INT_TYPE_INFO).print();
@@ -90,10 +90,9 @@ public class TypeFillTest extends StreamingMultipleProgramsTestBase {
map.print();
try {
- map.returns("String");
+ map.returns(String.class);
fail();
- } catch (Exception e) {
- }
+ } catch (Exception ignored) {}
}
@@ -101,14 +100,10 @@ public class TypeFillTest extends StreamingMultipleProgramsTestBase {
private static final long serialVersionUID = 1L;
@Override
- public void run(SourceContext<T> ctx) throws Exception {
-
- }
+ public void run(SourceContext<T> ctx) throws Exception {}
@Override
- public void cancel() {
-
- }
+ public void cancel() {}
}
private class TestMap<T, O> implements MapFunction<T, O> {
@@ -120,8 +115,7 @@ public class TypeFillTest extends StreamingMultipleProgramsTestBase {
private class TestFlatMap<T, O> implements FlatMapFunction<T, O> {
@Override
- public void flatMap(T value, Collector<O> out) throws Exception {
- }
+ public void flatMap(T value, Collector<O> out) throws Exception {}
}
private class TestCoMap<IN1, IN2, OUT> implements CoMapFunction<IN1, IN2, OUT> {
@@ -141,12 +135,10 @@ public class TypeFillTest extends StreamingMultipleProgramsTestBase {
private class TestCoFlatMap<IN1, IN2, OUT> implements CoFlatMapFunction<IN1, IN2, OUT> {
@Override
- public void flatMap1(IN1 value, Collector<OUT> out) throws Exception {
- }
+ public void flatMap1(IN1 value, Collector<OUT> out) throws Exception {}
@Override
- public void flatMap2(IN2 value, Collector<OUT> out) throws Exception {
- }
+ public void flatMap2(IN2 value, Collector<OUT> out) throws Exception {}
}
}