[
https://issues.apache.org/jira/browse/FLINK-7666?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16218302#comment-16218302
]
ASF GitHub Bot commented on FLINK-7666:
---------------------------------------
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/4900#discussion_r146801477
--- Diff:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/ProcessingTimeService.java
---
@@ -73,14 +73,19 @@
/**
* This method puts the service into a state where it does not register
new timers, but
* returns for each call to {@link #registerTimer(long,
ProcessingTimeCallback)} only a "mock" future.
- * Furthermore, the method clears all not yet started timers, and
awaits the completion
- * of currently executing timers.
+ * Furthermore, the method clears all not yet started timers.
*
* <p>This method can be used to cleanly shut down the timer service.
The using components
* will not notice that the service is shut down (as for example via
exceptions when registering
* a new timer), but the service will simply not fire any timer any
more.
*/
- public abstract void quiesceAndAwaitPending() throws
InterruptedException;
+ public abstract void quiesce() throws InterruptedException;
+
+ /**
+ * This method can be used after calling {@link #quiesce()}, and awaits
the completion
+ * of currently executing timers.
+ */
+ public abstract void awaitPendingAfterShutdown() throws
InterruptedException;
--- End diff --
nit: maybe `awaitPendingAfterQuiesce()` or only `awaitPending()`
> ContinuousFileReaderOperator swallows chained watermarks
> --------------------------------------------------------
>
> Key: FLINK-7666
> URL: https://issues.apache.org/jira/browse/FLINK-7666
> Project: Flink
> Issue Type: Improvement
> Components: Streaming Connectors
> Affects Versions: 1.3.2
> Reporter: Ufuk Celebi
> Assignee: Kostas Kloudas
> Priority: Blocker
> Fix For: 1.4.0
>
>
> I use event time and read from a (finite) file. I assign watermarks right
> after the {{ContinuousFileReaderOperator}} with parallelism 1.
> {code}
> env
> .readFile(new TextInputFormat(...), ...)
> .setParallelism(1)
> .assignTimestampsAndWatermarks(...)
> .setParallelism(1)
> .map()...
> {code}
> The watermarks I assign never progress through the pipeline.
> I can work around this by inserting a {{shuffle()}} after the file reader or
> starting a new chain at the assigner:
> {code}
> env
> .readFile(new TextInputFormat(...), ...)
> .setParallelism(1)
> .shuffle()
> .assignTimestampsAndWatermarks(...)
> .setParallelism(1)
> .map()...
> {code}
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
