You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by manuzhang <gi...@git.apache.org> on 2016/10/21 04:50:19 UTC
[GitHub] flink pull request #2671: [FLINK-4862] fix Timer register in ContinuousEvent...
GitHub user manuzhang opened a pull request:
https://github.com/apache/flink/pull/2671
[FLINK-4862] fix Timer register in ContinuousEventTimeTrigger
Thanks for contributing to Apache Flink. Before you open your pull request, please take the following check list into consideration.
If your changes take all of the items into account, feel free to open your pull request. For more information and/or questions please refer to the [How To Contribute guide](http://flink.apache.org/how-to-contribute.html).
In addition to going through the list, please provide a meaningful description of your changes.
- [ ] General
- The pull request references the related JIRA issue ("[FLINK-XXX] Jira title text")
- The pull request addresses only one issue
- Each commit in the PR has a meaningful commit message (including the JIRA id)
- [ ] Documentation
- Documentation has been added for new functionality
- Old documentation affected by the pull request has been updated
- JavaDoc for public methods has been added
- [ ] Tests & Build
- Functionality added by the pull request is covered by tests
- `mvn clean verify` has been executed successfully locally or a Travis build has passed
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/manuzhang/flink fix_merge_window
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/2671.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 #2671
----
commit b2370946357044250511e25fce5078812ad22c82
Author: manuzhang <ow...@gmail.com>
Date: 2016-10-20T07:06:01Z
[FLINK-4862] fix Timer register in ContinuousEventTimeTrigger
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2671: [FLINK-4862] fix Timer register in ContinuousEvent...
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2671#discussion_r84464167
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java ---
@@ -88,9 +96,13 @@ public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) t
@Override
public void clear(W window, TriggerContext ctx) throws Exception {
ReducingState<Long> fireTimestamp = ctx.getPartitionedState(stateDesc);
- long timestamp = fireTimestamp.get();
- ctx.deleteEventTimeTimer(timestamp);
- fireTimestamp.clear();
+ Long timestamp = fireTimestamp.get();
+ if (timestamp != null) {
+ ctx.deleteEventTimeTimer(timestamp);
+ fireTimestamp.clear();
+ } else if (cachedFireTimestamp != null){
+ ctx.deleteEventTimeTimer(cachedFireTimestamp);
+ }
--- End diff --
The above `else if` block is not correct because there is only one instance of the trigger which is reused for each Window. Hence the abstraction using the state descriptor to retrieve the appropriate state.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2671: [FLINK-4862] fix Timer register in ContinuousEventTimeTri...
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on the issue:
https://github.com/apache/flink/pull/2671
Thanks for having a look @aljoscha. Thanks again for the PR @manuzhang. Merging.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2671: [FLINK-4862] fix Timer register in ContinuousEvent...
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2671#discussion_r84464275
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java ---
@@ -45,6 +45,12 @@
private final ReducingStateDescriptor<Long> stateDesc =
new ReducingStateDescriptor<>("fire-time", new Min(), LongSerializer.INSTANCE);
+ /**
+ * Used to preserve the fire timestamp before merge such that
+ * the corresponding timer could be cleared after merge
+ */
+ private Long cachedFireTimestamp = null;
--- End diff --
This doesn't work because there is only one `Trigger` instance and this will potentially be overwritten by many Windows being merged.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2671: [FLINK-4862] fix Timer register in ContinuousEvent...
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2671#discussion_r84465826
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java ---
@@ -88,9 +96,13 @@ public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) t
@Override
public void clear(W window, TriggerContext ctx) throws Exception {
ReducingState<Long> fireTimestamp = ctx.getPartitionedState(stateDesc);
- long timestamp = fireTimestamp.get();
- ctx.deleteEventTimeTimer(timestamp);
- fireTimestamp.clear();
+ Long timestamp = fireTimestamp.get();
+ if (timestamp != null) {
+ ctx.deleteEventTimeTimer(timestamp);
+ fireTimestamp.clear();
--- End diff --
The above looks good.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2671: [FLINK-4862] fix Timer register in ContinuousEvent...
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2671#discussion_r84689568
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java ---
@@ -99,8 +111,12 @@ public boolean canMerge() {
}
@Override
- public TriggerResult onMerge(W window, OnMergeContext ctx) {
+ public TriggerResult onMerge(W window, OnMergeContext ctx) throws Exception {
ctx.mergePartitionedState(stateDesc);
+ Long nextFireTimestamp = ctx.getPartitionedState(stateDesc).get();
+ if (nextFireTimestamp != null) {
+ ctx.registerEventTimeTimer(nextFireTimestamp);
+ }
--- End diff --
Yes, you're right. It is actually handled correctly in `EventTimeTrigger` but not for the continuous trigger.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2671: [FLINK-4862] fix Timer register in ContinuousEvent...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/2671
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2671: [FLINK-4862] fix Timer register in ContinuousEventTimeTri...
Posted by manuzhang <gi...@git.apache.org>.
Github user manuzhang commented on the issue:
https://github.com/apache/flink/pull/2671
@mxm updated. Thanks for teaching me more about the internals. Ignoring the old Timer make things much simpler actually.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2671: [FLINK-4862] fix Timer register in ContinuousEventTimeTri...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/2671
@mxm @manuzhang Yes, the timers will be ignored because there is a check in `onEventTime()`.
The fix looks good! Will you go ahead and merge, @mxm?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2671: [FLINK-4862] fix Timer register in ContinuousEvent...
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2671#discussion_r84463868
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java ---
@@ -99,8 +111,12 @@ public boolean canMerge() {
}
@Override
- public TriggerResult onMerge(W window, OnMergeContext ctx) {
+ public TriggerResult onMerge(W window, OnMergeContext ctx) throws Exception {
ctx.mergePartitionedState(stateDesc);
+ Long nextFireTimestamp = ctx.getPartitionedState(stateDesc).get();
+ if (nextFireTimestamp != null) {
+ ctx.registerEventTimeTimer(nextFireTimestamp);
+ }
--- End diff --
+1
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2671: [FLINK-4862] fix Timer register in ContinuousEvent...
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2671#discussion_r84465211
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java ---
@@ -88,9 +96,13 @@ public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) t
@Override
public void clear(W window, TriggerContext ctx) throws Exception {
ReducingState<Long> fireTimestamp = ctx.getPartitionedState(stateDesc);
- long timestamp = fireTimestamp.get();
- ctx.deleteEventTimeTimer(timestamp);
- fireTimestamp.clear();
+ Long timestamp = fireTimestamp.get();
+ if (timestamp != null) {
+ ctx.deleteEventTimeTimer(timestamp);
+ fireTimestamp.clear();
+ } else if (cachedFireTimestamp != null){
+ ctx.deleteEventTimeTimer(cachedFireTimestamp);
+ }
--- End diff --
I think we're fine with not doing anything when `timestamp == null`. The old timer won't influence the newly merged window.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---