{"action":"created","issue":{"url":"https://api.github.com/repos/apache/flink/issues/906","labels_url":"https://api.github.com/repos/apache/flink/issues/906/labels{/name}","comments_url":"https://api.github.com/repos/apache/flink/issues/906/comments","events_url":"https://api.github.com/repos/apache/flink/issues/906/events","html_url":"https://github.com/apache/flink/pull/906","id":94765389,"number":906,"title":"[FLINK-1967] Introduce (Event)time in Streaming","user":{"login":"aljoscha","id":68551,"avatar_url":"https://avatars.githubusercontent.com/u/68551?v=3","gravatar_id":"","url":"https://api.github.com/users/aljoscha","html_url":"https://github.com/aljoscha","followers_url":"https://api.github.com/users/aljoscha/followers","following_url":"https://api.github.com/users/aljoscha/following{/other_user}","gists_url":"https://api.github.com/users/aljoscha/gists{/gist_id}","starred_url":"https://api.github.com/users/aljoscha/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/aljoscha/subscriptions","organizations_url":"https://api.github.com/users/aljoscha/orgs","repos_url":"https://api.github.com/users/aljoscha/repos","events_url":"https://api.github.com/users/aljoscha/events{/privacy}","received_events_url":"https://api.github.com/users/aljoscha/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":4,"created_at":"2015-07-13T16:55:32Z","updated_at":"2015-07-14T08:44:21Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/flink/pulls/906","html_url":"https://github.com/apache/flink/pull/906","diff_url":"https://github.com/apache/flink/pull/906.diff","patch_url":"https://github.com/apache/flink/pull/906.patch"},"body":"## [FLINK-1967] Introduce (Event)time in Streaming\r\n    \r\nThis introduces an additional timestamp field in StreamRecord. When using a\r\nSourceFunction and an auto-timestamp interval is set using the\r\nExecutionConfig, the timestamp is automatically set to System.currentTimeMillis()\r\nupon element emission. The timestamp can be manually set using an\r\nEventTimeSourceFunction.\r\n\r\nThis also changes the signature of the StreamOperators. They now get\r\na StreamRecord instead of the unwrapped value. This is necessary for\r\nthem to access the timestamp. Before, the StreamTask would unwrap the\r\nvalue from the StreamRecord, now this is moved one layer higher.\r\n\r\nThis also introduces watermarks to keep track of processing. At\r\na configurable interval the sources will emit watermarks that signify\r\nthat no records with a lower timestamp will be emitted in the future by\r\nthis source. The timestamps are broadcast, stream inputs wait for watermark\r\nevents on all input channels and forward the watermark to the\r\nStreamOperator once the watermark advances on all inputs. Operators are\r\nresponsible for forwarding the watermark once they know that no elements\r\nwith a previous timestamp will be emitted (this is mostly relevant for\r\nbuffering operations such as windows). Right now, all operators simply\r\nforward the watermark they receive.\r\n\r\nWhen using an EventTimeSourceFunction the system does not\r\nautomatically emit timestamps, the user is required to manually emit\r\nwatermarks using the SourceContext.\r\n\r\nNo watermarks will be emitted unless\r\nExecutionConfig.setAutoWatermarkInterval is used to set a watermark\r\ninterval.\r\n\r\nThis commit contains fixes for other stuff that was discovered while\r\nimplementing the feature. See Jira issue numbers and descriptions below.\r\n\r\n## [FLINK-2290/2295] Fix CoReader Event Handling and Checkpointing\r\n\r\nThis changes CoReader (now CoStreamingRecordReader) to reuse\r\nUnionGate for the input multiplexing. This way it will not lock in on\r\none specific input side and read events from both input sides.\r\n\r\nThis also enables an event listener for checkpoint barriers so that the\r\nTwoInputTask now reacts to those and correctly forwards them.\r\n\r\nThen, this adds CoStreamCheckpointintITCase to verify that checkpointing\r\nworks in topologies with TwoInputStreamTasks.\r\n\r\nThis also adds tests for OneInputStreamTask and TwoInputStreamTask\r\nthat check whether:\r\n - whether open()/close() of RichFunctions are correctly\r\n   called\r\n - Watermarks are correctly forwarded\r\n - Supersteps/checkpoint barriers are correctly forwarded and the\r\n   blocking of inputs works correctly\r\n\r\n## Add proper tests for Stream Operators\r\n\r\nThese test whether:\r\n - open()/close() on RichFunctions are called\r\n - Timestamps of emitted elements match the timestamp of the input\r\n   element\r\n - Watermarks are correctly forwarded\r\n\r\n## [FLINK-2301] Fix Behaviour of BarrierBuffer and add Tests\r\n\r\nBefore, a CheckpointBarrier from a more recent Checkpoint would also\r\ntrigger unblocking while waiting on an older CheckpointBarrier. Now,\r\na CheckpointBarrier from a newer checkpoint will unblock all channels\r\nand start a new wait on the new Checkpoint.\r\n\r\nThe tests for OneInputStreamTask and TwoInputStreamTask check whether\r\nthe buffer behaves correctly when receiving CheckpointBarriers from more\r\nrecent checkpoints while still waiting on an older CheckpointBarrier."},"comment":{"url":"https://api.github.com/repos/apache/flink/issues/comments/121168918","html_url":"https://github.com/apache/flink/pull/906#issuecomment-121168918","issue_url":"https://api.github.com/repos/apache/flink/issues/906","id":121168918,"user":{"login":"aljoscha","id":68551,"avatar_url":"https://avatars.githubusercontent.com/u/68551?v=3","gravatar_id":"","url":"https://api.github.com/users/aljoscha","html_url":"https://github.com/aljoscha","followers_url":"https://api.github.com/users/aljoscha/followers","following_url":"https://api.github.com/users/aljoscha/following{/other_user}","gists_url":"https://api.github.com/users/aljoscha/gists{/gist_id}","starred_url":"https://api.github.com/users/aljoscha/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/aljoscha/subscriptions","organizations_url":"https://api.github.com/users/aljoscha/orgs","repos_url":"https://api.github.com/users/aljoscha/repos","events_url":"https://api.github.com/users/aljoscha/events{/privacy}","received_events_url":"https://api.github.com/users/aljoscha/received_events","type":"User","site_admin":false},"created_at":"2015-07-14T08:44:21Z","updated_at":"2015-07-14T08:44:21Z","body":"The job I ran was source -> repartition -> sink, where the source is generating tuples in an infinite loop.\r\n\r\nThe gist of the data is this (I'm rerunning some of them with longer runtimes to remove outliers):\r\n - Introducing timestamp for each record that is set at source incurs about a 15 % performance hit in these tests \r\n - Watermarks have no additional overhead once the interval is higher than 100 sec (I didn't test with lower intervals, such as 50 sec, but I will do these today.)"}}