You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by rm...@apache.org on 2016/02/24 22:02:00 UTC
[3/4] flink git commit: [FLINK-3435] [streaming] Proparly separate
IngestionTime and EventTime
[FLINK-3435] [streaming] Proparly separate IngestionTime and EventTime
- IngestionTime now only auto-generates watermarks
- EventTime does not auto-generate ingestion watermarks
This also removes the "ExecutionConfig.areTimestampsEnabled()" flag.
This closes #1699
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ceb64248
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ceb64248
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ceb64248
Branch: refs/heads/master
Commit: ceb64248daab04b01977ff02516696e4398d656e
Parents: 004eb00
Author: Stephan Ewen <se...@apache.org>
Authored: Mon Feb 22 17:58:07 2016 +0100
Committer: Robert Metzger <rm...@apache.org>
Committed: Wed Feb 24 22:00:41 2016 +0100
----------------------------------------------------------------------
.../flink/storm/wrappers/BoltWrapper.java | 2 +-
.../storm/wrappers/MergedInputsBoltWrapper.java | 2 +-
.../flink/storm/wrappers/BoltWrapperTest.java | 6 +-
.../flink/api/common/ExecutionConfig.java | 39 ---
.../streaming/examples/join/WindowJoin.java | 2 +-
.../ml/IncrementalLearningSkeleton.java | 5 +-
.../examples/windowing/SessionWindowing.java | 4 +-
.../examples/windowing/TopSpeedWindowing.java | 2 +-
.../scala/examples/join/WindowJoin.scala | 2 +-
.../java/org/apache/flink/cep/CEPITCase.java | 12 +-
.../connectors/kafka/FlinkKafkaConsumer08.java | 4 +-
.../connectors/kafka/FlinkKafkaConsumer09.java | 4 +-
.../flink/streaming/api/TimeCharacteristic.java | 6 +-
.../streaming/api/datastream/DataStream.java | 8 +-
.../environment/StreamExecutionEnvironment.java | 2 -
.../functions/AscendingTimestampExtractor.java | 16 +-
.../AssignerWithPeriodicWatermarks.java | 13 +-
.../AssignerWithPunctuatedWatermarks.java | 11 +-
.../api/functions/IngestionTimeExtractor.java | 50 ++++
.../api/functions/TimestampAssigner.java | 10 +-
.../source/EventTimeSourceFunction.java | 42 ---
.../source/RichEventTimeSourceFunction.java | 49 ----
.../api/functions/source/SourceFunction.java | 4 +-
.../flink/streaming/api/graph/StreamConfig.java | 16 ++
.../flink/streaming/api/graph/StreamGraph.java | 13 +-
.../api/graph/StreamingJobGraphGenerator.java | 7 +-
.../streaming/api/operators/StreamFlatMap.java | 2 +-
.../streaming/api/operators/StreamSource.java | 236 ++++++++---------
.../api/operators/TimestampedCollector.java | 25 +-
.../api/operators/co/CoStreamFlatMap.java | 4 +-
.../windowing/assigners/SlidingTimeWindows.java | 25 +-
.../assigners/TumblingTimeWindows.java | 16 +-
.../api/windowing/assigners/WindowAssigner.java | 5 +-
...TimestampsAndPeriodicWatermarksOperator.java | 18 +-
...mestampsAndPunctuatedWatermarksOperator.java | 26 +-
...ractAlignedProcessingTimeWindowOperator.java | 4 +-
.../EvictingNonKeyedWindowOperator.java | 2 +-
.../windowing/EvictingWindowOperator.java | 2 +-
.../windowing/NonKeyedWindowOperator.java | 2 +-
.../operators/windowing/WindowOperator.java | 11 +-
.../MultiplexingStreamRecordSerializer.java | 135 ++++++----
.../runtime/streamrecord/StreamRecord.java | 108 ++++++--
.../streamrecord/StreamRecordSerializer.java | 11 +-
.../runtime/tasks/OneInputStreamTask.java | 2 +-
.../streaming/runtime/tasks/OperatorChain.java | 5 +-
.../runtime/tasks/SourceStreamTask.java | 51 +---
.../runtime/tasks/StreamIterationHead.java | 2 +-
.../streaming/runtime/tasks/StreamTask.java | 6 +
.../runtime/tasks/TwoInputStreamTask.java | 2 +-
.../flink/streaming/api/CoStreamITCase.java | 116 +++++++++
.../flink/streaming/api/CoStreamTest.java | 132 ----------
.../AscendingTimestampExtractorTest.java | 13 +-
.../functions/IngestionTimeExtractorTest.java | 65 +++++
.../source/FileMonitoringFunctionTest.java | 60 +++--
.../graph/StreamingJobGraphGeneratorTest.java | 2 +-
...stampsAndPeriodicWatermarksOperatorTest.java | 49 +++-
...ampsAndPunctuatedWatermarksOperatorTest.java | 46 +++-
.../operators/windowing/CoGroupJoinITCase.java | 10 +-
.../operators/windowing/WindowFoldITCase.java | 6 +-
.../MultiplexingStreamRecordSerializerTest.java | 108 ++++++++
.../StreamRecordSerializerTest.java | 18 ++
.../runtime/streamrecord/StreamRecordTest.java | 158 ++++++++++++
.../runtime/tasks/StreamTaskTestHarness.java | 4 +-
.../runtime/tasks/StreamTaskTimerITCase.java | 61 +++--
.../streaming/timestamp/TimestampITCase.java | 255 ++++++++-----------
.../streaming/util/CollectingSourceContext.java | 65 +++++
.../apache/flink/streaming/util/MockOutput.java | 3 +-
.../streaming/util/SourceFunctionUtil.java | 20 +-
.../flink/streaming/api/scala/DataStream.scala | 8 +-
.../streaming/api/scala/CoGroupJoinITCase.scala | 9 +-
.../streaming/api/scala/WindowFoldITCase.scala | 7 +-
.../EventTimeAllWindowCheckpointingITCase.java | 4 +-
.../EventTimeWindowCheckpointingITCase.java | 10 +-
73 files changed, 1367 insertions(+), 893 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltWrapper.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltWrapper.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltWrapper.java
index ef09a3a..8ac820e 100644
--- a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltWrapper.java
+++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/BoltWrapper.java
@@ -297,7 +297,7 @@ public class BoltWrapper<IN, OUT> extends AbstractStreamOperator<OUT> implements
@Override
public void processElement(final StreamRecord<IN> element) throws Exception {
- this.flinkCollector.setTimestamp(element.getTimestamp());
+ this.flinkCollector.setTimestamp(element);
IN value = element.getValue();
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/MergedInputsBoltWrapper.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/MergedInputsBoltWrapper.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/MergedInputsBoltWrapper.java
index 8682962..89defde 100644
--- a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/MergedInputsBoltWrapper.java
+++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/MergedInputsBoltWrapper.java
@@ -118,7 +118,7 @@ public final class MergedInputsBoltWrapper<IN, OUT> extends BoltWrapper<StormTup
@Override
public void processElement(final StreamRecord<StormTuple<IN>> element) throws Exception {
- this.flinkCollector.setTimestamp(element.getTimestamp());
+ this.flinkCollector.setTimestamp(element);
this.bolt.execute(element.getValue());
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltWrapperTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltWrapperTest.java b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltWrapperTest.java
index 1a583d4..e65dc45 100644
--- a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltWrapperTest.java
+++ b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltWrapperTest.java
@@ -64,7 +64,7 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@RunWith(PowerMockRunner.class)
-@PrepareForTest({StreamRecordSerializer.class, WrapperSetupHelper.class})
+@PrepareForTest({StreamRecordSerializer.class, WrapperSetupHelper.class, StreamRecord.class})
public class BoltWrapperTest extends AbstractTest {
@Test(expected = IllegalArgumentException.class)
@@ -197,7 +197,7 @@ public class BoltWrapperTest extends AbstractTest {
splitRecord.value = new Tuple1<Integer>(2);
}
wrapper.processElement(record);
- verify(output).collect(new StreamRecord<SplitStreamType>(splitRecord, 0));
+ verify(output).collect(new StreamRecord<SplitStreamType>(splitRecord));
if (rawOutType2) {
splitRecord.streamId = "stream2";
@@ -207,7 +207,7 @@ public class BoltWrapperTest extends AbstractTest {
splitRecord.value = new Tuple1<Integer>(3);
}
wrapper.processElement(record);
- verify(output, times(2)).collect(new StreamRecord<SplitStreamType>(splitRecord, 0));
+ verify(output, times(2)).collect(new StreamRecord<SplitStreamType>(splitRecord));
}
@SuppressWarnings("unchecked")
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
index a592680..ffd30e9 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
@@ -168,50 +168,11 @@ public class ExecutionConfig implements Serializable {
*/
@PublicEvolving
public ExecutionConfig setAutoWatermarkInterval(long interval) {
- enableTimestamps();
this.autoWatermarkInterval = interval;
return this;
}
/**
- * Enables streaming timestamps. When this is enabled all records that are emitted
- * from a source have a timestamp attached. This is required if a topology contains
- * operations that rely on watermarks and timestamps to perform operations, such as
- * event-time windows.
- *
- * <p>
- * This is automatically enabled if you enable automatic watermarks.
- *
- * @see #setAutoWatermarkInterval(long)
- */
- @PublicEvolving
- public ExecutionConfig enableTimestamps() {
- this.timestampsEnabled = true;
- return this;
- }
-
- /**
- * Disables streaming timestamps.
- *
- * @see #enableTimestamps()
- */
- @PublicEvolving
- public ExecutionConfig disableTimestamps() {
- this.timestampsEnabled = false;
- return this;
- }
-
- /**
- * Returns true when timestamps are enabled.
- *
- * @see #enableTimestamps()
- */
- @PublicEvolving
- public boolean areTimestampsEnabled() {
- return timestampsEnabled;
- }
-
- /**
* Returns the interval of the automatic watermark emission.
*
* @see #setAutoWatermarkInterval(long)
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/join/WindowJoin.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/join/WindowJoin.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/join/WindowJoin.java
index 216774b..8c52747 100644
--- a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/join/WindowJoin.java
+++ b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/join/WindowJoin.java
@@ -217,7 +217,7 @@ public class WindowJoin {
private static class MyTimestampExtractor extends AscendingTimestampExtractor<Tuple3<Long, String, Integer>> {
@Override
- public long extractAscendingTimestamp(Tuple3<Long, String, Integer> element, long currentTimestamp) {
+ public long extractAscendingTimestamp(Tuple3<Long, String, Integer> element) {
return element.f0;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/ml/IncrementalLearningSkeleton.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/ml/IncrementalLearningSkeleton.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/ml/IncrementalLearningSkeleton.java
index 4c1c40f..acbc5d6 100644
--- a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/ml/IncrementalLearningSkeleton.java
+++ b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/ml/IncrementalLearningSkeleton.java
@@ -25,6 +25,7 @@ import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
+import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
@@ -158,8 +159,8 @@ public class IncrementalLearningSkeleton {
}
@Override
- public long checkAndGetNextWatermark(Integer lastElement, long extractedTimestamp) {
- return counter - 1;
+ public Watermark checkAndGetNextWatermark(Integer lastElement, long extractedTimestamp) {
+ return new Watermark(counter - 1);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.java
index 50c31a3..16dbdfa 100644
--- a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.java
+++ b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.java
@@ -24,7 +24,7 @@ import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.source.EventTimeSourceFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
@@ -63,7 +63,7 @@ public class SessionWindowing {
input.add(new Tuple3<>("c", 11L, 1));
DataStream<Tuple3<String, Long, Integer>> source = env
- .addSource(new EventTimeSourceFunction<Tuple3<String,Long,Integer>>() {
+ .addSource(new SourceFunction<Tuple3<String,Long,Integer>>() {
private static final long serialVersionUID = 1L;
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowing.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowing.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowing.java
index 595286b..8f0eb47 100644
--- a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowing.java
+++ b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowing.java
@@ -166,7 +166,7 @@ public class TopSpeedWindowing {
private static final long serialVersionUID = 1L;
@Override
- public long extractAscendingTimestamp(Tuple4<Integer, Integer, Double, Long> element, long previous) {
+ public long extractAscendingTimestamp(Tuple4<Integer, Integer, Double, Long> element) {
return element.f3;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples/join/WindowJoin.scala
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples/join/WindowJoin.scala b/flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples/join/WindowJoin.scala
index fcc763d..81f12dc 100644
--- a/flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples/join/WindowJoin.scala
+++ b/flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples/join/WindowJoin.scala
@@ -46,7 +46,7 @@ object WindowJoin {
println("Usage: WindowJoin --grades <path> --salaries <path> --output <path>")
val env = StreamExecutionEnvironment.getExecutionEnvironment
- env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+ env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
env.getConfig.setGlobalJobParameters(params)
// Create streams for grades and salaries by mapping the inputs to the corresponding objects
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/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 8f0aa5d..37fa9d9 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
@@ -27,8 +27,8 @@ import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
+import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase;
import org.junit.After;
@@ -222,9 +222,8 @@ public class CEPITCase extends StreamingMultipleProgramsTestBase {
}
@Override
- public long checkAndGetNextWatermark(Tuple2<Event, Long> lastElement,
- long extractedTimestamp) {
- return lastElement.f1 - 5;
+ public Watermark checkAndGetNextWatermark(Tuple2<Event, Long> lastElement, long extractedTimestamp) {
+ return new Watermark(lastElement.f1 - 5);
}
}).map(new MapFunction<Tuple2<Event, Long>, Event>() {
@@ -308,9 +307,8 @@ public class CEPITCase extends StreamingMultipleProgramsTestBase {
}
@Override
- public long checkAndGetNextWatermark(Tuple2<Event, Long> lastElement,
- long extractedTimestamp) {
- return lastElement.f1 - 5;
+ public Watermark checkAndGetNextWatermark(Tuple2<Event, Long> lastElement, long extractedTimestamp) {
+ return new Watermark(lastElement.f1 - 5);
}
}).map(new MapFunction<Tuple2<Event, Long>, Event>() {
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java b/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java
index a738c4c..865fe36 100644
--- a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java
+++ b/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java
@@ -328,9 +328,7 @@ public class FlinkKafkaConsumer08<T> extends FlinkKafkaConsumerBase<T> {
else {
// this source never completes, so emit a Long.MAX_VALUE watermark
// to not block watermark forwarding
- if (getRuntimeContext().getExecutionConfig().areTimestampsEnabled()) {
- sourceContext.emitWatermark(new Watermark(Long.MAX_VALUE));
- }
+ sourceContext.emitWatermark(new Watermark(Long.MAX_VALUE));
final Object waitLock = new Object();
while (running) {
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java b/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java
index cdc483a..62ba3c4 100644
--- a/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java
+++ b/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java
@@ -338,9 +338,7 @@ public class FlinkKafkaConsumer09<T> extends FlinkKafkaConsumerBase<T> {
} else {
// this source never completes, so emit a Long.MAX_VALUE watermark
// to not block watermark forwarding
- if (getRuntimeContext().getExecutionConfig().areTimestampsEnabled()) {
- sourceContext.emitWatermark(new Watermark(Long.MAX_VALUE));
- }
+ sourceContext.emitWatermark(new Watermark(Long.MAX_VALUE));
final Object waitLock = new Object();
this.waitThread = Thread.currentThread();
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/TimeCharacteristic.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/TimeCharacteristic.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/TimeCharacteristic.java
index 44175e8..df2e4b9 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/TimeCharacteristic.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/TimeCharacteristic.java
@@ -59,9 +59,9 @@ public enum TimeCharacteristic {
* Event time means that the time of each individual element in the stream (also called event)
* is determined by the event's individual custom timestamp. These timestamps either exist in the
* elements from before they entered the Flink streaming dataflow, or are user-assigned at the sources.
- * The big implication of this is that elements arrive in the sources and in all operators generally
- * out of order, meaning that elements with earlier timestamps may arrive after elements with
- * later timestamps.
+ * The big implication of this is that it allows for elements to arrive in the sources and in
+ * all operators out of order, meaning that elements with earlier timestamps may arrive after
+ * elements with later timestamps.
* <p>
* Operators that window or order data with respect to event time must buffer data until they can
* be sure that all timestamps for a certain time interval have been received. This is handled by
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
index fa24e18..ddba7d6 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
@@ -768,17 +768,17 @@ public class DataStream<T> {
return transform("Timestamps/Watermarks", getTransformation().getOutputType(), operator)
.setParallelism(inputParallelism);
}
-
+
/**
* Assigns timestamps to the elements in the data stream and periodically creates
* watermarks to signal event time progress.
*
* <p>This method creates watermarks based purely on stream elements. For each element
* that is handled via {@link AssignerWithPunctuatedWatermarks#extractTimestamp(Object, long)},
- * the {@link AssignerWithPunctuatedWatermarks#checkAndGetNextWatermark(Object, long)}
+ * the {@link AssignerWithPunctuatedWatermarks#checkAndGetNextWatermark(Object, long)}
* method is called, and a new watermark is emitted, if the returned watermark value is
* non-negative and greater than the previous watermark.
- *
+ *
* <p>This method is useful when the data stream embeds watermark elements, or certain elements
* carry a marker that can be used to determine the current event time watermark.
* This operation gives the programmer full control over the watermark generation. Users
@@ -794,7 +794,7 @@ public class DataStream<T> {
*
* @see AssignerWithPunctuatedWatermarks
* @see AssignerWithPeriodicWatermarks
- * @see #assignTimestampsAndWatermarks(AssignerWithPeriodicWatermarks)
+ * @see #assignTimestampsAndWatermarks(AssignerWithPeriodicWatermarks)
*/
public SingleOutputStreamOperator<T, ?> assignTimestampsAndWatermarks(
AssignerWithPunctuatedWatermarks<T> timestampAndWatermarkAssigner) {
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
index 371ce77..4c1c265 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
@@ -596,10 +596,8 @@ public abstract class StreamExecutionEnvironment {
public void setStreamTimeCharacteristic(TimeCharacteristic characteristic) {
this.timeCharacteristic = requireNonNull(characteristic);
if (characteristic == TimeCharacteristic.ProcessingTime) {
- getConfig().disableTimestamps();
getConfig().setAutoWatermarkInterval(0);
} else {
- getConfig().enableTimestamps();
getConfig().setAutoWatermarkInterval(200);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AscendingTimestampExtractor.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AscendingTimestampExtractor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AscendingTimestampExtractor.java
index 60216b6..f2468d0 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AscendingTimestampExtractor.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AscendingTimestampExtractor.java
@@ -20,6 +20,7 @@ package org.apache.flink.streaming.api.functions;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.streaming.api.watermark.Watermark;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,21 +38,20 @@ public abstract class AscendingTimestampExtractor<T> implements AssignerWithPeri
private static final long serialVersionUID = 1L;
- /** The current timestamp */
- private long currentTimestamp = 0;
+ /** The current timestamp. */
+ private long currentTimestamp = Long.MIN_VALUE;
/** Handler that is called when timestamp monotony is violated */
private MonotonyViolationHandler violationHandler = new LoggingHandler();
+
/**
* Extracts the timestamp from the given element. The timestamp must be monotonically increasing.
*
* @param element The element that the timestamp is extracted from.
- * @param previousElementTimestamp The current internal timestamp of the element.
- *
* @return The new timestamp.
*/
- public abstract long extractAscendingTimestamp(T element, long previousElementTimestamp);
+ public abstract long extractAscendingTimestamp(T element);
/**
* Sets the handler for violations to the ascending timestamp order.
@@ -68,7 +68,7 @@ public abstract class AscendingTimestampExtractor<T> implements AssignerWithPeri
@Override
public final long extractTimestamp(T element, long elementPrevTimestamp) {
- final long newTimestamp = extractAscendingTimestamp(element, elementPrevTimestamp);
+ final long newTimestamp = extractAscendingTimestamp(element);
if (newTimestamp >= this.currentTimestamp) {
this.currentTimestamp = newTimestamp;
return newTimestamp;
@@ -79,8 +79,8 @@ public abstract class AscendingTimestampExtractor<T> implements AssignerWithPeri
}
@Override
- public final long getCurrentWatermark() {
- return currentTimestamp - 1;
+ public final Watermark getCurrentWatermark() {
+ return new Watermark(currentTimestamp == Long.MIN_VALUE ? Long.MIN_VALUE : currentTimestamp - 1);
}
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPeriodicWatermarks.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPeriodicWatermarks.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPeriodicWatermarks.java
index 821dd94..0c37896 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPeriodicWatermarks.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPeriodicWatermarks.java
@@ -19,6 +19,7 @@
package org.apache.flink.streaming.api.functions;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.streaming.api.watermark.Watermark;
/**
* The {@code AssignerWithPeriodicWatermarks} assigns event time timestamps to elements,
@@ -50,11 +51,13 @@ public interface AssignerWithPeriodicWatermarks<T> extends TimestampAssigner<T>
/**
* Returns the current watermark. This method is periodically called by the
- * system to retrieve the current watermark.
+ * system to retrieve the current watermark. The method may return null to
+ * indicate that no new Watermark is available.
*
- * <p>The current watermark will be emitted only if it is larger than the previously
- * emitted watermark. If the current watermark is still identical to the previous
- * one, no progress in event time has happened since the previous call to this method.
+ * <p>The returned watermark will be emitted only if it is non-null and larger
+ * than the previously emitted watermark. If the current watermark is still
+ * identical to the previous one, no progress in event time has happened since
+ * the previous call to this method.
*
* <p>If this method returns a value that is smaller than the previously returned watermark,
* then the implementation does not properly handle the event stream timestamps.
@@ -67,5 +70,5 @@ public interface AssignerWithPeriodicWatermarks<T> extends TimestampAssigner<T>
* @see org.apache.flink.streaming.api.watermark.Watermark
* @see ExecutionConfig#getAutoWatermarkInterval()
*/
- long getCurrentWatermark();
+ Watermark getCurrentWatermark();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPunctuatedWatermarks.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPunctuatedWatermarks.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPunctuatedWatermarks.java
index 39f3bb7..42579e3 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPunctuatedWatermarks.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPunctuatedWatermarks.java
@@ -18,6 +18,8 @@
package org.apache.flink.streaming.api.functions;
+import org.apache.flink.streaming.api.watermark.Watermark;
+
/**
* The {@code AssignerWithPunctuatedWatermarks} assigns event time timestamps to elements,
* and generates low watermarks that signal event time progress within the stream.
@@ -40,8 +42,8 @@ package org.apache.flink.streaming.api.functions;
* return element.getSequenceTimestamp();
* }
*
- * public long checkAndGetNextWatermark(MyElement lastElement, long extractedTimestamp) {
- * return lastElement.isEndOfSequence() ? extractedTimestamp : -1L;
+ * public Watermark checkAndGetNextWatermark(MyElement lastElement, long extractedTimestamp) {
+ * return lastElement.isEndOfSequence() ? new Watermark(extractedTimestamp) : null;
* }
* }
* }</pre>
@@ -70,8 +72,7 @@ public interface AssignerWithPunctuatedWatermarks<T> extends TimestampAssigner<T
* <p>For an example how to use this method, see the documentation of
* {@link AssignerWithPunctuatedWatermarks this class}.
*
- * @return A negative value, if no watermark should be emitted, positive value for
- * emitting this value as a watermark.
+ * @return {@code Null}, if no watermark should be emitted, or the next watermark to emit.
*/
- long checkAndGetNextWatermark(T lastElement, long extractedTimestamp);
+ Watermark checkAndGetNextWatermark(T lastElement, long extractedTimestamp);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/IngestionTimeExtractor.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/IngestionTimeExtractor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/IngestionTimeExtractor.java
new file mode 100644
index 0000000..c7c60d5
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/IngestionTimeExtractor.java
@@ -0,0 +1,50 @@
+/*
+ * 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.api.functions;
+
+import org.apache.flink.streaming.api.watermark.Watermark;
+
+/**
+ * A timestamp assigner that assigns timestamps based on the machine's wall clock.
+ *
+ * <p>If this assigner is used after a stream source, it realizes "ingestion time" semantics.
+ *
+ * @param <T> The elements that get timestamps assigned.
+ */
+public class IngestionTimeExtractor<T> implements AssignerWithPeriodicWatermarks<T> {
+ private static final long serialVersionUID = -4072216356049069301L;
+
+ private long maxTimestamp;
+
+ @Override
+ public long extractTimestamp(T element, long previousElementTimestamp) {
+ // make sure timestamps are monotonously increasing, even when the system clock re-syncs
+ final long now = Math.max(System.currentTimeMillis(), maxTimestamp);
+ maxTimestamp = now;
+ return now;
+ }
+
+ @Override
+ public Watermark getCurrentWatermark() {
+ // make sure timestamps are monotonously increasing, even when the system clock re-syncs
+ final long now = Math.max(System.currentTimeMillis(), maxTimestamp);
+ maxTimestamp = now;
+ return new Watermark(now - 1);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/TimestampAssigner.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/TimestampAssigner.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/TimestampAssigner.java
index 50fb0b4..5167fc8 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/TimestampAssigner.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/TimestampAssigner.java
@@ -28,10 +28,6 @@ import org.apache.flink.api.common.functions.Function;
* <p>Timestamps are represented in milliseconds since the Epoch
* (midnight, January 1, 1970 UTC).
*
- * <p>A timestamp assigner that assigns to each element a timestamp via
- * {@link System#currentTimeMillis()} effectively realizes "ingestion time"
- * semantics.
- *
* @param <T> The type of the elements to which this assigner assigns timestamps.
*/
public interface TimestampAssigner<T> extends Function {
@@ -39,10 +35,10 @@ public interface TimestampAssigner<T> extends Function {
/**
* Assigns a timestamp to an element, in milliseconds since the Epoch.
*
- * <p>The method gets the previously assigned timestamp of the element.
+ * <p>The method is passed the previously assigned timestamp of the element.
* That previous timestamp may have been assigned from a previous assigner,
- * by ingestion time, or be simply uninitialized. In the latter case, the
- * timestamp will be a negative value.
+ * by ingestion time. If the element did not carry a timestamp before, this value is
+ * {@code Long.MIN_VALUE}.
*
* @param element The element that the timestamp is wil be assigned to.
* @param previousElementTimestamp The previous internal timestamp of the element,
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/EventTimeSourceFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/EventTimeSourceFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/EventTimeSourceFunction.java
deleted file mode 100644
index 9b04ad9..0000000
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/EventTimeSourceFunction.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/**
- * 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.api.functions.source;
-
-
-import org.apache.flink.annotation.PublicEvolving;
-
-/**
- * A marker interface that must be implemented by {@link SourceFunction}s that emit elements with
- * timestamps. The {@link SourceFunction} can extract the timestamp from the data and attach it to
- * the element upon emission.
- *
- * <p>
- * Event-time sources must manually emit
- * {@link org.apache.flink.streaming.api.watermark.Watermark watermarks} to keep track of progress.
- * Automatic emission of watermarks will be suppressed if a source implements this interface.
- *
- * <p>
- * Elements must be emitted using
- * {@link SourceFunction.SourceContext#collectWithTimestamp(Object, long)}
- * and watermarks can be emitted using
- * {@link SourceFunction.SourceContext#emitWatermark(org.apache.flink.streaming.api.watermark.Watermark)}.
- *
- * @param <T> Type of the elements emitted by this source.
- */
-@PublicEvolving
-public interface EventTimeSourceFunction<T> extends SourceFunction<T> { }
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/RichEventTimeSourceFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/RichEventTimeSourceFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/RichEventTimeSourceFunction.java
deleted file mode 100644
index c56bfbb..0000000
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/RichEventTimeSourceFunction.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * 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.api.functions.source;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.functions.AbstractRichFunction;
-
-/**
- * Base class for implementing a parallel event-time data source that has access to context information
- * (via {@link #getRuntimeContext()}) and additional life-cycle methods
- * ({@link #open(org.apache.flink.configuration.Configuration)} and {@link #close()}.
- *
- * <p>
- * This class is useful when implementing parallel sources where different parallel subtasks
- * need to perform different work. Typical patterns for that are:
- * <ul>
- * <li>Use {@link #getRuntimeContext()} to obtain the runtime context.</li>
- * <li>Use {@link org.apache.flink.api.common.functions.RuntimeContext#getNumberOfParallelSubtasks()}
- * to determine the current parallelism. It is strongly encouraged to use this method, rather than
- * hard-wiring the parallelism, because the configured parallelism may change depending on
- * program configuration. The parallelism may also change after recovering failures, when fewer than
- * desired parallel worker as available.</li>
- * <li>Use {@link org.apache.flink.api.common.functions.RuntimeContext#getIndexOfThisSubtask()} to
- * determine which subtask the current instance of the function executes.</li>
- * </ul>
- *
- *
- * @param <OUT> The type of the records produced by this source.
- */
-@PublicEvolving
-public abstract class RichEventTimeSourceFunction<OUT> extends AbstractRichFunction implements EventTimeSourceFunction<OUT> {
-
- private static final long serialVersionUID = 1L;
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java
index c4139bd..1f4e1e4 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java
@@ -116,7 +116,7 @@ public interface SourceFunction<T> extends Function, Serializable {
* @param <T> The type of the elements produced by the source.
*/
@Public // Interface might be extended in the future with additional methods.
- public static interface SourceContext<T> {
+ interface SourceContext<T> {
/**
* Emits one element from the source. The result of {@link System#currentTimeMillis()} is set as
@@ -133,7 +133,7 @@ public interface SourceFunction<T> extends Function, Serializable {
* @param timestamp The timestamp in milliseconds
*/
@PublicEvolving
- public void collectWithTimestamp(T element, long timestamp);
+ void collectWithTimestamp(T element, long timestamp);
/**
* Emits the given {@link org.apache.flink.streaming.api.watermark.Watermark}.
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
index 1abb189..c9b1d38 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
@@ -29,8 +29,10 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.operators.util.CorruptConfigurationException;
import org.apache.flink.runtime.util.ClassLoaderUtil;
import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.runtime.state.AbstractStateBackend;
@@ -73,6 +75,7 @@ public class StreamConfig implements Serializable {
private static final String STATE_PARTITIONER = "statePartitioner";
private static final String STATE_KEY_SERIALIZER = "statekeyser";
+ private static final String TIME_CHARACTERISTIC = "timechar";
// ------------------------------------------------------------------------
// Default Values
@@ -108,6 +111,19 @@ public class StreamConfig implements Serializable {
return config.getInteger(VERTEX_NAME, -1);
}
+ public void setTimeCharacteristic(TimeCharacteristic characteristic) {
+ config.setInteger(TIME_CHARACTERISTIC, characteristic.ordinal());
+ }
+
+ public TimeCharacteristic getTimeCharacteristic() {
+ int ordinal = config.getInteger(TIME_CHARACTERISTIC, -1);
+ if (ordinal >= 0) {
+ return TimeCharacteristic.values()[ordinal];
+ } else {
+ throw new CorruptConfigurationException("time characteristic is not set");
+ }
+ }
+
public void setTypeSerializerIn1(TypeSerializer<?> serializer) {
setTypeSerializer(TYPE_SERIALIZER_IN_1, serializer);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
index abe1943..4be2874 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
@@ -74,7 +74,7 @@ public class StreamGraph extends StreamingPlan {
private String jobName = StreamExecutionEnvironment.DEFAULT_JOB_NAME;
- private final StreamExecutionEnvironment environemnt;
+ private final StreamExecutionEnvironment environment;
private final ExecutionConfig executionConfig;
private final CheckpointConfig checkpointConfig;
@@ -93,7 +93,7 @@ public class StreamGraph extends StreamingPlan {
public StreamGraph(StreamExecutionEnvironment environment) {
- this.environemnt = environment;
+ this.environment = environment;
this.executionConfig = environment.getConfig();
this.checkpointConfig = environment.getCheckpointConfig();
@@ -114,6 +114,11 @@ public class StreamGraph extends StreamingPlan {
sources = new HashSet<>();
sinks = new HashSet<>();
}
+
+
+ public StreamExecutionEnvironment getEnvironment() {
+ return environment;
+ }
public ExecutionConfig getExecutionConfig() {
return executionConfig;
@@ -251,7 +256,7 @@ public class StreamGraph extends StreamingPlan {
throw new RuntimeException("Duplicate vertexID " + vertexID);
}
- StreamNode vertex = new StreamNode(environemnt,
+ StreamNode vertex = new StreamNode(environment,
vertexID,
slotSharingGroup,
operatorObject,
@@ -572,7 +577,7 @@ public class StreamGraph extends StreamingPlan {
StreamingJobGraphGenerator jobgraphGenerator = new StreamingJobGraphGenerator(this);
- return jobgraphGenerator.createJobGraph(jobName);
+ return jobgraphGenerator.createJobGraph();
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
index e74efdf..da46424 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
@@ -20,7 +20,9 @@ package org.apache.flink.streaming.api.graph;
import com.google.common.hash.HashFunction;
import com.google.common.hash.Hasher;
import com.google.common.hash.Hashing;
+
import org.apache.commons.lang3.StringUtils;
+
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.operators.util.UserCodeObjectWrapper;
@@ -51,6 +53,7 @@ import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner;
import org.apache.flink.streaming.runtime.tasks.StreamIterationHead;
import org.apache.flink.streaming.runtime.tasks.StreamIterationTail;
import org.apache.flink.util.InstantiationUtil;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -108,7 +111,7 @@ public class StreamingJobGraphGenerator {
this.physicalEdgesInOrder = new ArrayList<>();
}
- public JobGraph createJobGraph(String jobName) {
+ public JobGraph createJobGraph() {
jobGraph = new JobGraph(streamGraph.getJobName());
// make sure that all vertices start immediately
@@ -323,6 +326,8 @@ public class StreamingJobGraphGenerator {
config.setNonChainedOutputs(nonChainableOutputs);
config.setChainedOutputs(chainableOutputs);
+ config.setTimeCharacteristic(streamGraph.getEnvironment().getStreamTimeCharacteristic());
+
final CheckpointConfig ceckpointCfg = streamGraph.getCheckpointConfig();
config.setStateBackend(streamGraph.getStateBackend());
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamFlatMap.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamFlatMap.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamFlatMap.java
index 3a1d537..ec7b713 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamFlatMap.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamFlatMap.java
@@ -44,7 +44,7 @@ public class StreamFlatMap<IN, OUT>
@Override
public void processElement(StreamRecord<IN> element) throws Exception {
- collector.setTimestamp(element.getTimestamp());
+ collector.setTimestamp(element);
userFunction.flatMap(element.getValue(), collector);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/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 ccf0edd..2470417 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
@@ -18,8 +18,7 @@
package org.apache.flink.streaming.api.operators;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.streaming.api.functions.source.EventTimeSourceFunction;
+import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -36,10 +35,11 @@ import java.util.concurrent.TimeUnit;
* @param <SRC> Type of the source function of this stream source operator
*/
@Internal
-public class StreamSource<OUT, SRC extends SourceFunction<OUT>> extends AbstractUdfStreamOperator<OUT, SRC>
- implements StreamOperator<OUT> {
+public class StreamSource<OUT, SRC extends SourceFunction<OUT>>
+ extends AbstractUdfStreamOperator<OUT, SRC> implements StreamOperator<OUT> {
private static final long serialVersionUID = 1L;
+
private transient SourceFunction.SourceContext<OUT> ctx;
public StreamSource(SRC sourceFunction) {
@@ -49,16 +49,21 @@ public class StreamSource<OUT, SRC extends SourceFunction<OUT>> extends Abstract
}
public void run(final Object lockingObject, final Output<StreamRecord<OUT>> collector) throws Exception {
- final ExecutionConfig executionConfig = getExecutionConfig();
+ final TimeCharacteristic timeCharacteristic = getOperatorConfig().getTimeCharacteristic();
- if (userFunction instanceof EventTimeSourceFunction) {
- ctx = new ManualWatermarkContext<OUT>(lockingObject, collector, getRuntimeContext().getExecutionConfig().areTimestampsEnabled());
- } else if (executionConfig.getAutoWatermarkInterval() > 0) {
- ctx = new AutomaticWatermarkContext<OUT>(lockingObject, collector, executionConfig);
- } else if (executionConfig.areTimestampsEnabled()) {
- ctx = new NonWatermarkContext<OUT>(lockingObject, collector);
- } else {
- ctx = new NonTimestampContext<OUT>(lockingObject, collector);
+ switch (timeCharacteristic) {
+ case EventTime:
+ ctx = new ManualWatermarkContext<>(this, lockingObject, collector);
+ break;
+ case IngestionTime:
+ ctx = new AutomaticWatermarkContext<>(this, lockingObject, collector,
+ getRuntimeContext().getExecutionConfig().getAutoWatermarkInterval());
+ break;
+ case ProcessingTime:
+ ctx = new NonTimestampContext<>(this, lockingObject, collector);
+ break;
+ default:
+ throw new Exception(String.valueOf(timeCharacteristic));
}
userFunction.run(ctx);
@@ -73,91 +78,51 @@ public class StreamSource<OUT, SRC extends SourceFunction<OUT>> extends Abstract
ctx.close();
}
}
-
- /**
- * {@link SourceFunction.SourceContext} to be used for sources that don't emit watermarks.
- * In addition to {@link NonWatermarkContext} this will also not attach timestamps to sources.
- * (Technically it will always set the timestamp to 0).
- */
- public static class NonTimestampContext<T> implements SourceFunction.SourceContext<T> {
-
- private final Object lockingObject;
- private final Output<StreamRecord<T>> output;
- private final StreamRecord<T> reuse;
-
- public NonTimestampContext(Object lockingObjectParam, Output<StreamRecord<T>> outputParam) {
- this.lockingObject = lockingObjectParam;
- this.output = outputParam;
- this.reuse = new StreamRecord<T>(null);
- }
-
- @Override
- public void collect(T element) {
- output.collect(reuse.replace(element, 0));
- }
-
- @Override
- public void collectWithTimestamp(T element, long timestamp) {
- throw new UnsupportedOperationException("Automatic-Timestamp sources cannot emit" +
- " elements with a timestamp. See interface EventTimeSourceFunction" +
- " if you want to manually assign timestamps to elements.");
- }
-
- @Override
- public void emitWatermark(Watermark mark) {
- throw new UnsupportedOperationException("Automatic-Timestamp sources cannot emit" +
- " elements with a timestamp. See interface EventTimeSourceFunction" +
- " if you want to manually assign timestamps to elements.");
- }
-
- @Override
- public Object getCheckpointLock() {
- return lockingObject;
- }
-
- @Override
- public void close() {}
+
+ void checkAsyncException() {
+ getContainingTask().checkTimerException();
}
+ // ------------------------------------------------------------------------
+ // Source contexts for various stream time characteristics
+ // ------------------------------------------------------------------------
+
/**
- * {@link SourceFunction.SourceContext} to be used for sources that don't emit watermarks.
+ * A source context that attached {@code -1} as a timestamp to all records, and that
+ * does not forward watermarks.
*/
- public static class NonWatermarkContext<T> implements SourceFunction.SourceContext<T> {
+ public static class NonTimestampContext<T> implements SourceFunction.SourceContext<T> {
+ private final StreamSource<?, ?> owner;
private final Object lockingObject;
private final Output<StreamRecord<T>> output;
private final StreamRecord<T> reuse;
- public NonWatermarkContext(Object lockingObjectParam, Output<StreamRecord<T>> outputParam) {
- this.lockingObject = lockingObjectParam;
- this.output = outputParam;
+ public NonTimestampContext(StreamSource<?, ?> owner, Object lockingObject, Output<StreamRecord<T>> output) {
+ this.owner = owner;
+ this.lockingObject = lockingObject;
+ this.output = output;
this.reuse = new StreamRecord<T>(null);
}
@Override
public void collect(T element) {
- long currentTime = System.currentTimeMillis();
- output.collect(reuse.replace(element, currentTime));
+ owner.checkAsyncException();
+ synchronized (lockingObject) {
+ output.collect(reuse.replace(element));
+ }
}
@Override
public void collectWithTimestamp(T element, long timestamp) {
- throw new UnsupportedOperationException("Automatic-Timestamp sources cannot emit" +
- " elements with a timestamp. See interface EventTimeSourceFunction" +
- " if you want to manually assign timestamps to elements.");
+ // ignore the timestamp
+ collect(element);
}
@Override
public void emitWatermark(Watermark mark) {
- if (mark.getTimestamp() == Long.MAX_VALUE) {
- // allow it since this is the special end-watermark that for example the Kafka
- // source emits
- output.emitWatermark(mark);
- } else {
- throw new UnsupportedOperationException("Automatic-Timestamp sources cannot emit" +
- " elements with a timestamp. See interface EventTimeSourceFunction" +
- " if you want to manually assign timestamps to elements.");
- }
+ owner.checkAsyncException();
+ // do nothing else
}
@Override
@@ -168,86 +133,100 @@ public class StreamSource<OUT, SRC extends SourceFunction<OUT>> extends Abstract
@Override
public void close() {}
}
-
+
/**
* {@link SourceFunction.SourceContext} to be used for sources with automatic timestamps
* and watermark emission.
*/
public static class AutomaticWatermarkContext<T> implements SourceFunction.SourceContext<T> {
+ private final StreamSource<?, ?> owner;
+ private final Object lockingObject;
+ private final Output<StreamRecord<T>> output;
+ private final StreamRecord<T> reuse;
+
private final ScheduledExecutorService scheduleExecutor;
private final ScheduledFuture<?> watermarkTimer;
private final long watermarkInterval;
- private final Object lockingObject;
- private final Output<StreamRecord<T>> output;
- private final StreamRecord<T> reuse;
+ private volatile long nextWatermarkTime;
- private volatile long lastWatermarkTime;
+ public AutomaticWatermarkContext(
+ final StreamSource<?, ?> owner,
+ final Object lockingObjectParam,
+ final Output<StreamRecord<T>> outputParam,
+ final long watermarkInterval) {
+
+ if (watermarkInterval < 1L) {
+ throw new IllegalArgumentException("The watermark interval cannot be smaller than one.");
+ }
- public AutomaticWatermarkContext(Object lockingObjectParam,
- Output<StreamRecord<T>> outputParam,
- ExecutionConfig executionConfig) {
+ this.owner = owner;
this.lockingObject = lockingObjectParam;
this.output = outputParam;
+ this.watermarkInterval = watermarkInterval;
this.reuse = new StreamRecord<T>(null);
+
+ this.scheduleExecutor = Executors.newScheduledThreadPool(1);
- watermarkInterval = executionConfig.getAutoWatermarkInterval();
-
- scheduleExecutor = Executors.newScheduledThreadPool(1);
-
- watermarkTimer = scheduleExecutor.scheduleAtFixedRate(new Runnable() {
+ this.watermarkTimer = scheduleExecutor.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
- long currentTime = System.currentTimeMillis();
- // align the watermarks across all machines. this will ensure that we
- // don't have watermarks that creep along at different intervals because
- // the machine clocks are out of sync
- long watermarkTime = currentTime - (currentTime % watermarkInterval);
- if (currentTime > watermarkTime && watermarkTime - lastWatermarkTime >= watermarkInterval) {
- synchronized (lockingObject) {
- if (currentTime > watermarkTime && watermarkTime - lastWatermarkTime >= watermarkInterval) {
- output.emitWatermark(new Watermark(watermarkTime));
- lastWatermarkTime = watermarkTime;
+ final long currentTime = System.currentTimeMillis();
+
+ if (currentTime > nextWatermarkTime) {
+ // align the watermarks across all machines. this will ensure that we
+ // don't have watermarks that creep along at different intervals because
+ // the machine clocks are out of sync
+ final long watermarkTime = currentTime - (currentTime % watermarkInterval);
+
+ synchronized (lockingObjectParam) {
+ if (currentTime > nextWatermarkTime) {
+ outputParam.emitWatermark(new Watermark(watermarkTime));
+ nextWatermarkTime += watermarkInterval;
}
}
}
}
}, 0, watermarkInterval, TimeUnit.MILLISECONDS);
-
}
@Override
public void collect(T element) {
+ owner.checkAsyncException();
+
synchronized (lockingObject) {
- long currentTime = System.currentTimeMillis();
+ final long currentTime = System.currentTimeMillis();
output.collect(reuse.replace(element, currentTime));
-
- long watermarkTime = currentTime - (currentTime % watermarkInterval);
- if (currentTime > watermarkTime && watermarkTime - lastWatermarkTime >= watermarkInterval) {
+
+ if (currentTime > nextWatermarkTime) {
+ // in case we jumped some watermarks, recompute the next watermark time
+ final long watermarkTime = currentTime - (currentTime % watermarkInterval);
+ nextWatermarkTime = watermarkTime + watermarkInterval;
output.emitWatermark(new Watermark(watermarkTime));
- lastWatermarkTime = watermarkTime;
}
}
}
@Override
public void collectWithTimestamp(T element, long timestamp) {
- throw new UnsupportedOperationException("Automatic-Timestamp sources cannot emit" +
- " elements with a timestamp. See interface EventTimeSourceFunction" +
- " if you want to manually assign timestamps to elements.");
+ collect(element);
}
@Override
public void emitWatermark(Watermark mark) {
+ owner.checkAsyncException();
+
if (mark.getTimestamp() == Long.MAX_VALUE) {
- // allow it since this is the special end-watermark that for example the Kafka
- // source emits
- output.emitWatermark(mark);
- } else {
- throw new UnsupportedOperationException("Automatic-Timestamp sources cannot emit" +
- " elements with a timestamp. See interface EventTimeSourceFunction" +
- " if you want to manually assign timestamps to elements.");
+ // allow it since this is the special end-watermark that for example the Kafka source emits
+ synchronized (lockingObject) {
+ nextWatermarkTime = Long.MAX_VALUE;
+ output.emitWatermark(mark);
+ }
+
+ // we can shutdown the timer now, no watermarks will be needed any more
+ watermarkTimer.cancel(true);
+ scheduleExecutor.shutdownNow();
}
}
@@ -264,31 +243,40 @@ public class StreamSource<OUT, SRC extends SourceFunction<OUT>> extends Abstract
}
/**
- * {@link SourceFunction.SourceContext} to be used for sources with manual timestamp
- * assignment and manual watermark emission.
+ * A SourceContext for event time. Sources may directly attach timestamps and generate
+ * watermarks, but if records are emitted without timestamps, no timetamps are automatically
+ * generated and attached. The records will simply have no timestamp in that case.
+ *
+ * Streaming topologies can use timestamp assigner functions to override the timestamps
+ * assigned here.
*/
public static class ManualWatermarkContext<T> implements SourceFunction.SourceContext<T> {
+ private final StreamSource<?, ?> owner;
private final Object lockingObject;
private final Output<StreamRecord<T>> output;
private final StreamRecord<T> reuse;
- private final boolean watermarkMultiplexingEnabled;
- public ManualWatermarkContext(Object lockingObject, Output<StreamRecord<T>> output, boolean watermarkMultiplexingEnabled) {
+ public ManualWatermarkContext(StreamSource<?, ?> owner, Object lockingObject, Output<StreamRecord<T>> output) {
+ this.owner = owner;
this.lockingObject = lockingObject;
this.output = output;
this.reuse = new StreamRecord<T>(null);
- this.watermarkMultiplexingEnabled = watermarkMultiplexingEnabled;
}
@Override
public void collect(T element) {
- throw new UnsupportedOperationException("Manual-Timestamp sources can only emit" +
- " elements with a timestamp. Please use collectWithTimestamp().");
+ owner.checkAsyncException();
+
+ synchronized (lockingObject) {
+ output.collect(reuse.replace(element));
+ }
}
@Override
public void collectWithTimestamp(T element, long timestamp) {
+ owner.checkAsyncException();
+
synchronized (lockingObject) {
output.collect(reuse.replace(element, timestamp));
}
@@ -296,7 +284,9 @@ public class StreamSource<OUT, SRC extends SourceFunction<OUT>> extends Abstract
@Override
public void emitWatermark(Watermark mark) {
- if (watermarkMultiplexingEnabled) {
+ owner.checkAsyncException();
+
+ synchronized (lockingObject) {
output.emitWatermark(mark);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/TimestampedCollector.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/TimestampedCollector.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/TimestampedCollector.java
index 05b65c5..56fa14d 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/TimestampedCollector.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/TimestampedCollector.java
@@ -29,7 +29,7 @@ import org.apache.flink.util.Collector;
* would set the timestamp of the incoming
* {@link org.apache.flink.streaming.runtime.streamrecord.StreamRecord} here.
*
- * @param <T> The type of the elments that can be emitted.
+ * @param <T> The type of the elements that can be emitted.
*/
@Internal
public class TimestampedCollector<T> implements Collector<T> {
@@ -37,9 +37,6 @@ public class TimestampedCollector<T> implements Collector<T> {
private final Output<StreamRecord<T>> output;
private final StreamRecord<T> reuse;
-
- private long timestamp;
-
/**
* Creates a new {@link TimestampedCollector} that wraps the given {@link Output}.
@@ -51,17 +48,19 @@ public class TimestampedCollector<T> implements Collector<T> {
@Override
public void collect(T record) {
- output.collect(reuse.replace(record, timestamp));
+ output.collect(reuse.replace(record));
+ }
+
+ public void setTimestamp(StreamRecord<?> timestampBase) {
+ if (timestampBase.hasTimestamp()) {
+ reuse.setTimestamp(timestampBase.getTimestamp());
+ } else {
+ reuse.eraseTimestamp();
+ }
}
- /**
- * Sets the timestamp (long milliseconds) that is attached to elements that get emitted using
- * {@link #collect(Object)}
- *
- * @param timestamp The timestamp in milliseconds
- */
- public void setTimestamp(long timestamp) {
- this.timestamp = timestamp;
+ public void setAbsoluteTimestamp(long timestamp) {
+ reuse.setTimestamp(timestamp);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/co/CoStreamFlatMap.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/co/CoStreamFlatMap.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/co/CoStreamFlatMap.java
index 679187c..580a860 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/co/CoStreamFlatMap.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/co/CoStreamFlatMap.java
@@ -52,14 +52,14 @@ public class CoStreamFlatMap<IN1, IN2, OUT>
@Override
public void processElement1(StreamRecord<IN1> element) throws Exception {
- collector.setTimestamp(element.getTimestamp());
+ collector.setTimestamp(element);
userFunction.flatMap1(element.getValue(), collector);
}
@Override
public void processElement2(StreamRecord<IN2> element) throws Exception {
- collector.setTimestamp(element.getTimestamp());
+ collector.setTimestamp(element);
userFunction.flatMap2(element.getValue(), collector);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/SlidingTimeWindows.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/SlidingTimeWindows.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/SlidingTimeWindows.java
index e658c41..900e254 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/SlidingTimeWindows.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/SlidingTimeWindows.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.windowing.assigners;
import org.apache.flink.annotation.PublicEvolving;
@@ -58,14 +59,20 @@ public class SlidingTimeWindows extends WindowAssigner<Object, TimeWindow> {
@Override
public Collection<TimeWindow> assignWindows(Object element, long timestamp) {
- List<TimeWindow> windows = new ArrayList<>((int) (size / slide));
- long lastStart = timestamp - timestamp % slide;
- for (long start = lastStart;
- start > timestamp - size;
- start -= slide) {
- windows.add(new TimeWindow(start, start + size));
+ if (timestamp > Long.MIN_VALUE) {
+ List<TimeWindow> windows = new ArrayList<>((int) (size / slide));
+ long lastStart = timestamp - timestamp % slide;
+ for (long start = lastStart;
+ start > timestamp - size;
+ start -= slide) {
+ windows.add(new TimeWindow(start, start + size));
+ }
+ return windows;
+ } else {
+ throw new RuntimeException("Record has Long.MIN_VALUE timestamp (= no timestamp marker). " +
+ "Is the time characteristic set to 'ProcessingTime', or did you forget to call " +
+ "'DataStream.assignTimestampsAndWatermarks(...)'?");
}
- return windows;
}
public long getSize() {
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/TumblingTimeWindows.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/TumblingTimeWindows.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/TumblingTimeWindows.java
index f4f8c28..2b97691 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/TumblingTimeWindows.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/TumblingTimeWindows.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.windowing.assigners;
import org.apache.flink.annotation.PublicEvolving;
@@ -54,8 +55,15 @@ public class TumblingTimeWindows extends WindowAssigner<Object, TimeWindow> {
@Override
public Collection<TimeWindow> assignWindows(Object element, long timestamp) {
- long start = timestamp - (timestamp % size);
- return Collections.singletonList(new TimeWindow(start, start + size));
+ if (timestamp > Long.MIN_VALUE) {
+ // Long.MIN_VALUE is currently assigned when no timestamp is present
+ long start = timestamp - (timestamp % size);
+ return Collections.singletonList(new TimeWindow(start, start + size));
+ } else {
+ throw new RuntimeException("Record has Long.MIN_VALUE timestamp (= no timestamp marker). " +
+ "Is the time characteristic set to 'ProcessingTime', or did you forget to call " +
+ "'DataStream.assignTimestampsAndWatermarks(...)'?");
+ }
}
public long getSize() {
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/WindowAssigner.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/WindowAssigner.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/WindowAssigner.java
index c788924..683ed42 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/WindowAssigner.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/WindowAssigner.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.windowing.assigners;
import org.apache.flink.annotation.PublicEvolving;
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndPeriodicWatermarksOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndPeriodicWatermarksOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndPeriodicWatermarksOperator.java
index c7a14c7..fbd46ef 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndPeriodicWatermarksOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndPeriodicWatermarksOperator.java
@@ -35,7 +35,7 @@ public class TimestampsAndPeriodicWatermarksOperator<T>
implements OneInputStreamOperator<T, T>, Triggerable {
private static final long serialVersionUID = 1L;
-
+
private transient long watermarkInterval;
private transient long currentWatermark;
@@ -43,14 +43,14 @@ public class TimestampsAndPeriodicWatermarksOperator<T>
public TimestampsAndPeriodicWatermarksOperator(AssignerWithPeriodicWatermarks<T> assigner) {
super(assigner);
- chainingStrategy = ChainingStrategy.ALWAYS;
+ this.chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
public void open() throws Exception {
super.open();
- currentWatermark = -1L;
+ currentWatermark = Long.MIN_VALUE;
watermarkInterval = getExecutionConfig().getAutoWatermarkInterval();
if (watermarkInterval > 0) {
@@ -60,18 +60,20 @@ public class TimestampsAndPeriodicWatermarksOperator<T>
@Override
public void processElement(StreamRecord<T> element) throws Exception {
- long newTimestamp = userFunction.extractTimestamp(element.getValue(), element.getTimestamp());
+ final long newTimestamp = userFunction.extractTimestamp(element.getValue(),
+ element.hasTimestamp() ? element.getTimestamp() : Long.MIN_VALUE);
+
output.collect(element.replace(element.getValue(), newTimestamp));
}
@Override
public void trigger(long timestamp) throws Exception {
// register next timer
- long newWatermark = userFunction.getCurrentWatermark();
- if (newWatermark > currentWatermark) {
- currentWatermark = newWatermark;
+ Watermark newWatermark = userFunction.getCurrentWatermark();
+ if (newWatermark != null && newWatermark.getTimestamp() > currentWatermark) {
+ currentWatermark = newWatermark.getTimestamp();
// emit watermark
- output.emitWatermark(new Watermark(currentWatermark));
+ output.emitWatermark(newWatermark);
}
registerTimer(System.currentTimeMillis() + watermarkInterval, this);
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndPunctuatedWatermarksOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndPunctuatedWatermarksOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndPunctuatedWatermarksOperator.java
index 75649eb..ac85b8a 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndPunctuatedWatermarksOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndPunctuatedWatermarksOperator.java
@@ -36,30 +36,26 @@ public class TimestampsAndPunctuatedWatermarksOperator<T>
private static final long serialVersionUID = 1L;
- private transient long currentWatermark;
+ private long currentWatermark = Long.MIN_VALUE;
+
-
public TimestampsAndPunctuatedWatermarksOperator(AssignerWithPunctuatedWatermarks<T> assigner) {
super(assigner);
- chainingStrategy = ChainingStrategy.ALWAYS;
- }
-
- @Override
- public void open() throws Exception {
- super.open();
- currentWatermark = -1L;
+ this.chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
public void processElement(StreamRecord<T> element) throws Exception {
final T value = element.getValue();
- final long newTimestamp = userFunction.extractTimestamp(value, element.getTimestamp());
+ final long newTimestamp = userFunction.extractTimestamp(value,
+ element.hasTimestamp() ? element.getTimestamp() : Long.MIN_VALUE);
+
output.collect(element.replace(element.getValue(), newTimestamp));
-
- final long nextWatermark = userFunction.checkAndGetNextWatermark(value, newTimestamp);
- if (nextWatermark >= 0 && nextWatermark > currentWatermark) {
- currentWatermark = nextWatermark;
- output.emitWatermark(new Watermark(nextWatermark));
+
+ final Watermark nextWatermark = userFunction.checkAndGetNextWatermark(value, newTimestamp);
+ if (nextWatermark != null && nextWatermark.getTimestamp() > currentWatermark) {
+ currentWatermark = nextWatermark.getTimestamp();
+ output.emitWatermark(nextWatermark);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/AbstractAlignedProcessingTimeWindowOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/AbstractAlignedProcessingTimeWindowOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/AbstractAlignedProcessingTimeWindowOperator.java
index 6e00240..af341a6 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/AbstractAlignedProcessingTimeWindowOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/AbstractAlignedProcessingTimeWindowOperator.java
@@ -28,8 +28,8 @@ import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.runtime.state.StateHandle;
import org.apache.flink.runtime.util.MathUtils;
import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.TimestampedCollector;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
@@ -234,7 +234,7 @@ public abstract class AbstractAlignedProcessingTimeWindowOperator<KEY, IN, OUT,
}
private void computeWindow(long timestamp) throws Exception {
- out.setTimestamp(timestamp);
+ out.setAbsoluteTimestamp(timestamp);
panes.truncatePanes(numPanesPerWindow);
panes.evaluateWindow(out, new TimeWindow(timestamp, timestamp + windowSize), this);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingNonKeyedWindowOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingNonKeyedWindowOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingNonKeyedWindowOperator.java
index 3844c55..510ebb2 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingNonKeyedWindowOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingNonKeyedWindowOperator.java
@@ -59,7 +59,7 @@ public class EvictingNonKeyedWindowOperator<IN, OUT, W extends Window> extends N
@Override
@SuppressWarnings("unchecked, rawtypes")
protected void emitWindow(Context context) throws Exception {
- timestampedCollector.setTimestamp(context.window.maxTimestamp());
+ timestampedCollector.setAbsoluteTimestamp(context.window.maxTimestamp());
EvictingWindowBuffer<IN> windowBuffer = (EvictingWindowBuffer<IN>) context.windowBuffer;
int toEvict = 0;
http://git-wip-us.apache.org/repos/asf/flink/blob/ceb64248/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperator.java
index e9daa18..cfab3d5 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperator.java
@@ -104,7 +104,7 @@ public class EvictingWindowOperator<K, IN, OUT, W extends Window> extends Window
}
if (triggerResult.isFire()) {
- timestampedCollector.setTimestamp(window.maxTimestamp());
+ timestampedCollector.setAbsoluteTimestamp(window.maxTimestamp());
setKeyContext(key);