flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Tzu-Li (Gordon) Tai (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
Date Thu, 03 Nov 2016 06:53:58 GMT

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

Tzu-Li (Gordon) Tai commented on FLINK-4576:
--------------------------------------------

Back to analysis of the original problem:

I think the only problem that we were facing is that our own sources, i.e. Kafka and Kinesis
in particular, may have instances that initially are idle, but may eventually still have data
to read when partitions / shard counts scale up. Before, we were sending max value watermarks
from these idle subtasks downstream to ensure that downstream operators can still advance
their watermark.
However, due to FLINK-4022 and FLINK-4341, we don't really want to do that, because it breaks
the possibility of letting idle subtasks transparently accept new partitions / shards to read
from without messing up the watermarks downstream.

To solve this, I'm considering another approach:
- Have a new special watermark, a subclass of {{Watermark}}, that when a downstream operator
receive from one of it's input channels, the operator will mark that input channel as temporarily
"watermark idle", and will not take it into account when determining whether or not to advance
the timers.
- This special watermark is not passed on downstream. It basically stops at the operator that
directly receives it afterwards, to let it know that one of it's input channels will indefinitely
halt to send any watermarks and should not be accounted for when advancing watermarks.
- When an input channel that was previously marked as "watermark idle" sends out a normal
watermark, the behaviour falls back to like before.
- We basically only want sources to be sending this special watermark, and shouldn't be used
anywhere else.

There is of course the problem that user-assigned watermarks in the middle of the topology
can still mess up / not emit watermarks, but I think it is reasonable to have a contract with
user-provided timestamp / watermark assigners that their implementations should not cause
this.

[~aljoscha] what do you think?

> Low Watermark Service in JobManager for Streaming Sources
> ---------------------------------------------------------
>
>                 Key: FLINK-4576
>                 URL: https://issues.apache.org/jira/browse/FLINK-4576
>             Project: Flink
>          Issue Type: New Feature
>          Components: JobManager, Streaming, TaskManager
>            Reporter: Tzu-Li (Gordon) Tai
>            Assignee: Tzu-Li (Gordon) Tai
>            Priority: Blocker
>             Fix For: 1.2.0
>
>
> As per discussion in FLINK-4341 by [~aljoscha] and [~StephanEwen], we need a low watermark
service in the JobManager to support transparent resharding / partition discovery for our
Kafka and Kinesis consumers (and any future streaming connectors in general for which the
external system may elastically scale up and down independently of the parallelism of sources
in Flink). The main idea is to let source subtasks that don't emit their own watermarks (because
they currently don't have data partitions to consume) emit the low watermark across all subtasks,
instead of simply emitting a Long.MAX_VALUE watermark and forbidding them to be assigned partitions
in the future.
> The proposed implementation, from a high-level: a {{LowWatermarkCoordinator}} will be
added to execution graphs, periodically triggering only the source vertices with a {{RetrieveLowWatermark}}
message. The tasks reply to the JobManager through the actor gateway (or a new interface after
FLINK-4456 gets merged) with a {{ReplyLowWatermark}} message. When the coordinator collects
all low watermarks for a particular source vertex and determines the aggregated low watermark
for this round (accounting only values that are larger than the aggregated low watermark of
the last round), it sends a {{NotifyNewLowWatermark}} message to the source vertex's tasks.
> The messages will only be relevant to tasks that implement an internal {{LowWatermarkCooperatingTask}}
interface. For now, only {{SourceStreamTask}} should implement {{LowWatermarkCooperatingTask}}.
> Source functions should implement a public {{LowWatermarkListener}} interface if they
wish to get notified of the aggregated low watermarks across subtasks. Connectors like the
Kinesis consumer can choose to emit this watermark if the subtask currently does not have
any shards, so that downstream operators may still properly advance time windows (implementation
for this is tracked as a separate issue).
> Overall, the service will include -
> New messages between JobManager <-> TaskManager:
> {{RetrieveLowWatermark(jobId, jobVertexId, taskId, timestamp)}}
> {{ReplyLowWatermark(jobId, jobVertexId, taskId, currentLowWatermark)}}
> {{NotifyNewLowWatermark(jobId, jobVertexId, taskId, newLowWatermark, timestamp)}}
> New internal task interface {{LowWatermarkCooperatingTask}} in flink-runtime
> New public interface {{LowWatermarkListener}} in flink-streaming-java
> Might also need to extend {{SourceFunction.SourceContext}} to support retrieving the
current low watermark of sources.
> Any feedback for this is appreciated!



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

Mime
View raw message