[ 
https://issues.apache.org/jira/browse/FLINK-3674?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15585554#comment-15585554
 ] 

ASF GitHub Bot commented on FLINK-3674:
---------------------------------------

Github user StefanRRichter commented on a diff in the pull request:

    https://github.com/apache/flink/pull/2570#discussion_r83854971
  
    --- Diff: 
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/Triggerable.java
 ---
    @@ -15,38 +15,26 @@
      * See the License for the specific language governing permissions and
      * limitations under the License.
      */
    +package org.apache.flink.streaming.api.operators;
     
    -package org.apache.flink.streaming.runtime.operators.windowing;
    +import org.apache.flink.annotation.Internal;
     
    -import org.apache.flink.streaming.runtime.operators.Triggerable;
    -import org.apache.flink.streaming.runtime.tasks.TimeServiceProvider;
    -
    -import java.util.concurrent.ScheduledFuture;
    -
    -class NoOpTimerService extends TimeServiceProvider {
    -
    -   private volatile boolean terminated;
    -
    -   @Override
    -   public long getCurrentProcessingTime() {
    -           return System.currentTimeMillis();
    -   }
    -
    -   @Override
    -   public ScheduledFuture<?> registerTimer(long timestamp, Triggerable 
target) {
    -           return null;
    -   }
    -
    -   @Override
    -   public boolean isTerminated() {
    -           return terminated;
    -   }
    -
    -   @Override
    -   public void quiesceAndAwaitPending() {}
    -
    -   @Override
    -   public void shutdownService() {
    -           terminated = true;
    -   }
    +/**
    + * Interface for things that can be called by {@link InternalTimerService}.
    + *
    + * @param <K> Type of the keys to which timers are scoped.
    + * @param <N> Type of the namespace to which timers are scoped.
    + */
    +@Internal
    +public interface Triggerable<K, N> {
    --- End diff --
    
    I found this name a bit confusing, it could lead to confusion with the 
concept of Triggers. Furthermore, there is another class with the same simple 
name in a different package. Maybe this could be called `TimerCallback`?


> Add an interface for Time aware User Functions
> ----------------------------------------------
>
>                 Key: FLINK-3674
>                 URL: https://issues.apache.org/jira/browse/FLINK-3674
>             Project: Flink
>          Issue Type: New Feature
>          Components: Streaming
>    Affects Versions: 1.0.0
>            Reporter: Stephan Ewen
>            Assignee: Aljoscha Krettek
>
> I suggest to add an interface that UDFs can implement, which will let them be 
> notified upon watermark updates.
> Example usage:
> {code}
> public interface EventTimeFunction {
>     void onWatermark(Watermark watermark);
> }
> public class MyMapper implements MapFunction<String, String>, 
> EventTimeFunction {
>     private long currentEventTime = Long.MIN_VALUE;
>     public String map(String value) {
>         return value + " @ " + currentEventTime;
>     }
>     public void onWatermark(Watermark watermark) {
>         currentEventTime = watermark.getTimestamp();
>     }
> }
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Reply via email to