[GitHub] [flink] sunhaibotb commented on a change in pull request #10151: [FLINK-14231] Handle the processing-time timers before closing operator to properly support endInput

[GitBox <gi...@apache.org>]

sunhaibotb commented on a change in pull request #10151: [FLINK-14231] Handle the processing-time timers before closing operator to properly support endInput
URL: https://github.com/apache/flink/pull/10151#discussion_r347295130
 
 

 ##########
 File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/ProcessingTimeServiceImpl.java
 ##########
 @@ -39,11 +63,110 @@ public long getCurrentProcessingTime() {
 
 	@Override
 	public ScheduledFuture<?> registerTimer(long timestamp, ProcessingTimeCallback target) {
-		return timerService.registerTimer(timestamp, processingTimeCallbackWrapper.apply(target));
+		if (isQuiesced) {
 
 Review comment:
   This is to keep the same behavior as before. Look at `SystemProcessingTimeService#registerTimer` for the previous behavior:
   
   ```
   @Override
   public ScheduledFuture<?> registerTimer(long timestamp, ProcessingTimeCallback callback) {
       ......
       if (status == STATUS_QUIESCED) {
   	return new NeverCompleteFuture(delay);
       }
       ......
   }
   ```

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services