You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by StefanRRichter <gi...@git.apache.org> on 2018/05/29 14:10:26 UTC
[GitHub] flink pull request #6096: [FLINK-9440][streaming] Expose cancelation of time...
GitHub user StefanRRichter opened a pull request:
https://github.com/apache/flink/pull/6096
[FLINK-9440][streaming] Expose cancelation of timers through timer service interface
## What is the purpose of the change
This PR is enhancing PR #6077 and #6062. The purpose is to expose timer cancelation that has been optimized in the previously mentioned PRs to the user.
## Brief change log
Introduced `TimerService::deleteProcessingTimeTimer(long)` and `TimerService::deleteEventTimeTimer(long)`.
## Verifying this change
Existing tests in `KeyedProcessOperatorTest` and `KeyedCoProcessOperatorTest` have been enhanced to cover the new functionality.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
- The S3 file system connector: no)
## Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (JavaDocs)
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/StefanRRichter/flink expose-timer-delete
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/6096.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 #6096
----
commit fffc7ff6750509de2c97ea7ed44d2404669acd35
Author: Stefan Richter <s....@...>
Date: 2018-05-08T14:00:55Z
[FLINK-9423][state] Implement efficient deletes for heap-based timer service.
commit a8e0f2f6bf4343938819e2e9a3b7ea29e94d0fc0
Author: Stefan Richter <s....@...>
Date: 2018-05-23T15:39:47Z
review comments
commit 4fc37152905f80ba2c6db2d0fdc08273a751c34b
Author: Stefan Richter <s....@...>
Date: 2018-05-24T13:08:02Z
[FLINK-9436][state] Remove generic parameter namespace from InternalTimeServiceManager.
commit 3813ffd4eafb3884b6c0d7aa7f4aa155c943fb68
Author: Stefan Richter <s....@...>
Date: 2018-05-28T14:52:53Z
[FLINK-9440][streaming] Expose cancelation of timers through timer service interface.
----
---
[GitHub] flink issue #6096: [FLINK-9440][streaming] Expose cancelation of timers thro...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on the issue:
https://github.com/apache/flink/pull/6096
@tillrohrmann thanks for the fast review! Will merge.
---
[GitHub] flink issue #6096: [FLINK-9440][streaming] Expose cancelation of timers thro...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on the issue:
https://github.com/apache/flink/pull/6096
CC @aljoscha or @tillrohrmann
---
[GitHub] flink pull request #6096: [FLINK-9440][streaming] Expose cancelation of time...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/6096
---
[GitHub] flink pull request #6096: [FLINK-9440][streaming] Expose cancelation of time...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/6096#discussion_r192994475
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/ProcessOperator.java ---
@@ -114,12 +114,22 @@ public long currentWatermark() {
@Override
public void registerProcessingTimeTimer(long time) {
- throw new UnsupportedOperationException("Setting timers is only supported on a KeyedStream.");
+ throw new UnsupportedOperationException(UNSUPPORTED_REGISTER_TIMER_MSG);
}
@Override
public void registerEventTimeTimer(long time) {
- throw new UnsupportedOperationException("Setting timers is only supported on a KeyedStream.");
+ throw new UnsupportedOperationException(UNSUPPORTED_REGISTER_TIMER_MSG);
+ }
+
+ @Override
+ public void deleteProcessingTimeTimer(long time) {
+ throw new UnsupportedOperationException(UNSUPPORTED_DELETE_TIMER_MSG);
+ }
+
+ @Override
+ public void deleteEventTimeTimer(long time) {
+ throw new UnsupportedOperationException(UNSUPPORTED_DELETE_TIMER_MSG);
}
--- End diff --
The reason is that `ProcessFunction` can be used both on a keyed stream and a non-keyed stream. There is the newer `KeyedProcessFunction`, which should be the only function that exposes timers and state but for historic reasons it's currently like this. 😅
---
[GitHub] flink pull request #6096: [FLINK-9440][streaming] Expose cancelation of time...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6096#discussion_r192976292
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/ProcessOperator.java ---
@@ -114,12 +114,22 @@ public long currentWatermark() {
@Override
public void registerProcessingTimeTimer(long time) {
- throw new UnsupportedOperationException("Setting timers is only supported on a KeyedStream.");
+ throw new UnsupportedOperationException(UNSUPPORTED_REGISTER_TIMER_MSG);
}
@Override
public void registerEventTimeTimer(long time) {
- throw new UnsupportedOperationException("Setting timers is only supported on a KeyedStream.");
+ throw new UnsupportedOperationException(UNSUPPORTED_REGISTER_TIMER_MSG);
+ }
+
+ @Override
+ public void deleteProcessingTimeTimer(long time) {
+ throw new UnsupportedOperationException(UNSUPPORTED_DELETE_TIMER_MSG);
+ }
+
+ @Override
+ public void deleteEventTimeTimer(long time) {
+ throw new UnsupportedOperationException(UNSUPPORTED_DELETE_TIMER_MSG);
}
--- End diff --
I know it is out of scope, but why do we give access to the `TimerService` in a non-keyed context? Wouldn't it be better to follow the interface-segregation principle a bit more here and expose the timer service only to keyed operators? Is there a reason why we don't do it like that @aljoscha?
---
[GitHub] flink pull request #6096: [FLINK-9440][streaming] Expose cancelation of time...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6096#discussion_r193043954
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/ProcessOperator.java ---
@@ -114,12 +114,22 @@ public long currentWatermark() {
@Override
public void registerProcessingTimeTimer(long time) {
- throw new UnsupportedOperationException("Setting timers is only supported on a KeyedStream.");
+ throw new UnsupportedOperationException(UNSUPPORTED_REGISTER_TIMER_MSG);
}
@Override
public void registerEventTimeTimer(long time) {
- throw new UnsupportedOperationException("Setting timers is only supported on a KeyedStream.");
+ throw new UnsupportedOperationException(UNSUPPORTED_REGISTER_TIMER_MSG);
+ }
+
+ @Override
+ public void deleteProcessingTimeTimer(long time) {
+ throw new UnsupportedOperationException(UNSUPPORTED_DELETE_TIMER_MSG);
+ }
+
+ @Override
+ public void deleteEventTimeTimer(long time) {
+ throw new UnsupportedOperationException(UNSUPPORTED_DELETE_TIMER_MSG);
}
--- End diff --
https://issues.apache.org/jira/browse/FLINK-9529
---
[GitHub] flink pull request #6096: [FLINK-9440][streaming] Expose cancelation of time...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/6096#discussion_r193040105
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/ProcessOperator.java ---
@@ -114,12 +114,22 @@ public long currentWatermark() {
@Override
public void registerProcessingTimeTimer(long time) {
- throw new UnsupportedOperationException("Setting timers is only supported on a KeyedStream.");
+ throw new UnsupportedOperationException(UNSUPPORTED_REGISTER_TIMER_MSG);
}
@Override
public void registerEventTimeTimer(long time) {
- throw new UnsupportedOperationException("Setting timers is only supported on a KeyedStream.");
+ throw new UnsupportedOperationException(UNSUPPORTED_REGISTER_TIMER_MSG);
+ }
+
+ @Override
+ public void deleteProcessingTimeTimer(long time) {
+ throw new UnsupportedOperationException(UNSUPPORTED_DELETE_TIMER_MSG);
+ }
+
+ @Override
+ public void deleteEventTimeTimer(long time) {
+ throw new UnsupportedOperationException(UNSUPPORTED_DELETE_TIMER_MSG);
}
--- End diff --
may want to add open a JIRA now so we don't forget it.
---
[GitHub] flink pull request #6096: [FLINK-9440][streaming] Expose cancelation of time...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6096#discussion_r193039294
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/ProcessOperator.java ---
@@ -114,12 +114,22 @@ public long currentWatermark() {
@Override
public void registerProcessingTimeTimer(long time) {
- throw new UnsupportedOperationException("Setting timers is only supported on a KeyedStream.");
+ throw new UnsupportedOperationException(UNSUPPORTED_REGISTER_TIMER_MSG);
}
@Override
public void registerEventTimeTimer(long time) {
- throw new UnsupportedOperationException("Setting timers is only supported on a KeyedStream.");
+ throw new UnsupportedOperationException(UNSUPPORTED_REGISTER_TIMER_MSG);
+ }
+
+ @Override
+ public void deleteProcessingTimeTimer(long time) {
+ throw new UnsupportedOperationException(UNSUPPORTED_DELETE_TIMER_MSG);
+ }
+
+ @Override
+ public void deleteEventTimeTimer(long time) {
+ throw new UnsupportedOperationException(UNSUPPORTED_DELETE_TIMER_MSG);
}
--- End diff --
Alright, thanks for the clarification. I guess we could change it with Flink 2.0 because it would break the API to change it.
---