You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by bowenli86 <gi...@git.apache.org> on 2017/10/08 23:30:42 UTC
[GitHub] flink pull request #4786: [FLINK-7388][DataStream API] ProcessFunction.onTim...
GitHub user bowenli86 opened a pull request:
https://github.com/apache/flink/pull/4786
[FLINK-7388][DataStream API] ProcessFunction.onTimer() sets processing time as timestamp
## What is the purpose of the change
Before Flink 1.4.0, when called from a processing-time timer, the `ProcessFunction.onTimer()` method sets the current processing time as event-time timestamp. This behavior is very subtle and might not be noticed by users.
Well, it's harmful because processing-time timestamps are indeterministic and not aligned with watermarks. Besides, user-implemented logic depends on this wrong timestamp highly likely is unintendedly faulty.
So we've decided to fix it. Upon upgrading to 1.4.0, Flink jobs that are using this incorrect event-time timestamp will fail, and users should adapt their jobs to the correct logic.
## Brief change log
- not set a timestamp for processing-time timer
## Verifying this change
This change is already covered by existing tests
## Does this pull request potentially affect one of the following parts:
None
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (docs)
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/bowenli86/flink FLINK-7388
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/4786.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #4786
----
commit 1fec305efba8ad94b7d7e72a672186caaf40270e
Author: Bowen Li <bo...@gmail.com>
Date: 2017-10-08T22:57:51Z
[FLINK-7388] ProcessFunction.onTimer() sets processing time as timestamp
commit 7e9afde7519381adadf222febea9640afd41fa2a
Author: Bowen Li <bo...@gmail.com>
Date: 2017-10-08T23:26:54Z
add doc
----
---
[GitHub] flink issue #4786: [FLINK-7388][DataStream API] ProcessFunction.onTimer() se...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/4786
Thanks a lot, @bowenli86. I just merged this. 👍
Could you please close the PR?
---
[GitHub] flink pull request #4786: [FLINK-7388][DataStream API] ProcessFunction.onTim...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on a diff in the pull request:
https://github.com/apache/flink/pull/4786#discussion_r143517965
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/KeyedProcessOperator.java ---
@@ -79,7 +79,6 @@ public void onEventTime(InternalTimer<K, VoidNamespace> timer) throws Exception
@Override
public void onProcessingTime(InternalTimer<K, VoidNamespace> timer) throws Exception {
- collector.setAbsoluteTimestamp(timer.getTimestamp());
--- End diff --
@aljoscha Thanks for pointing this out. Fixed!
---
[GitHub] flink pull request #4786: [FLINK-7388][DataStream API] ProcessFunction.onTim...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/4786#discussion_r143502884
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/KeyedProcessOperator.java ---
@@ -79,7 +79,6 @@ public void onEventTime(InternalTimer<K, VoidNamespace> timer) throws Exception
@Override
public void onProcessingTime(InternalTimer<K, VoidNamespace> timer) throws Exception {
- collector.setAbsoluteTimestamp(timer.getTimestamp());
--- End diff --
This should call `eraseTimestamp()` because we might still have a timestamp set from processing some previous elements. Same for the other occurrences in the code.
---
[GitHub] flink pull request #4786: [FLINK-7388][DataStream API] ProcessFunction.onTim...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 closed the pull request at:
https://github.com/apache/flink/pull/4786
---
[GitHub] flink issue #4786: [FLINK-7388][DataStream API] ProcessFunction.onTimer() se...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on the issue:
https://github.com/apache/flink/pull/4786
Hi @aljoscha @fhueske , please let me know if you have any more comments
---
[GitHub] flink issue #4786: [FLINK-7388][DataStream API] ProcessFunction.onTimer() se...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the issue:
https://github.com/apache/flink/pull/4786
Yes, I think this behavior would be better.
Thanks, Fabian
---
[GitHub] flink issue #4786: [FLINK-7388][DataStream API] ProcessFunction.onTimer() se...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/4786
@fhueske is this what you had in mind?
---