[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15636243#comment-15636243
]
Tzu-Li (Gordon) Tai commented on FLINK-4576:
The re-opened tickets, split as separate tasks:
FLINK-5017 Introduce {{WatermarkStatus}} stream element (idle / active)
FLINK-5018 Make source idle timeout user configurable
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15636021#comment-15636021
]
Tzu-Li (Gordon) Tai commented on FLINK-4576:
Thanks for all the input, [~aljoscha] and [~StephanEwen]! I think we have a
solid idea and plan for the solution now.
Since the final discussed solution is completely different from what this
ticket originally proposed, I'll close this one as "won't fix" and open a new
one. I'll try to open a preview PR as soon as possible.
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15636003#comment-15636003
]
Tzu-Li (Gordon) Tai commented on FLINK-4576:
Oh, I see what you mean by timeout now. Yes, I think it's good to have a
user-configurable timeout value to start considering subtasks as idle, when no
data is read from assigned partitions.
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15635986#comment-15635986
]
Aljoscha Krettek commented on FLINK-4576:
-
Yes, that was my thinking as well.
The {{OperatorChain}} code is a bit of a mess. But yes, essentially
{{ChainingOutput}} directly forwards elements/watermarks/etc to the next
operator in the chain.
The logic with the idle watermarks will be somewhat tricky to get right:
normally, the watermark is the minimum of the input watermarks. Assume that an
operator has three inputs, one is idle, the others have watermark 10 and 15,
respectively. Now, the idle operator starts emitting again and the watermark
that it emits is 5. Our operator now has to ignore those watermarks until they
catch up to the current minimum watermark (which was 10 but could increase in
the meantime) because the watermark is not allowed to go backwards.
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15635990#comment-15635990
]
Aljoscha Krettek commented on FLINK-4576:
-
Without a timeout, when is a source considered idle? Only when it doesn't have
partitions assigned?
This is fine with me, but in that case users need to ensure that all partitions
get data, otherwise an existing partition might hold back the watermark if it
never gets any data.
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15635727#comment-15635727
]
Tzu-Li (Gordon) Tai commented on FLINK-4576:
If we let sources emit "watermark idle" and "watermark reactive" messages, do
we still need a user-defined idle timeout? The sources should guarantee that
there will be no stream elements (records / watermarks) in between a "watermark
idle" and "watermark reactivate."
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15635719#comment-15635719
]
Tzu-Li (Gordon) Tai commented on FLINK-4576:
Yes, that's true. In general, if all input channels of an operator has received
the "watermark idle" message, then itself should propagate a "watermark idle"
downstream. In the case of an operator which is in the middle of the operator
chain of a task, it should just pass on the "watermark idle" downstream. In the
case of a head operator of a chain which has multiple input channels, it only
passes a "watermark idle" message when all of its input channels has received
the message.
About that, I could use some help describing how the code works regarding
elements flowing through an operator chain of a single task. Specifically, if a
previous operator in the chain emits an watermark, this watermark will be used
to call the {{AbstractStreamOperator#processWatermark}} method of the operator
directly afterwards, correct? I'm struggling a bit trying to understand how the
{{ChainedOutput}} works in {{OperatorChain}}.
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15635687#comment-15635687
]
Aljoscha Krettek commented on FLINK-4576:
-
Scratch that, maybe we just need to add the idle detection logic to all places
that emit watermarks. Possibly with a user-defined idle timeout. This way, we
can have watermark emitters throughout the pipeline.
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15635684#comment-15635684
]
Aljoscha Krettek commented on FLINK-4576:
-
Scratch that, maybe we just need to add the idle detection logic to all places
that emit watermarks. Possibly with a user-defined idle timeout. This way, we
can have watermark emitters throughout the pipeline.
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15635640#comment-15635640
]
Aljoscha Krettek commented on FLINK-4576:
-
Yes, that could work.
It seems there has to be some coordination, at least in the case "source ->
timestamp extractor" where the timestamp extractor is not directly in the
source this extractor needs to realise that the input is idle and itself
forward the idle watermark, correct?
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15635107#comment-15635107
]
Tzu-Li (Gordon) Tai commented on FLINK-4576:
I think the general problem with a central watermark service to coordinate the
global watermark of an operator's instances is that its breaking the nature
that "event watermarks are derived from the data itself".
Addressing [~aljoscha]'s concerns:
I think adding also a "watermark reactivate" message that sources also send
should address some of the issues. The purpose of the "watermark reactivate"
message is to avoid naively determining elements that arrive after the
"watermark reactivate" message as late.
- For subtasks with partitions that actually don't produce new elements for a
while, I think the source's itself should implement a threshold that when
reached, a "watermark idle" message should be emitted downstream. As soon as
the source subtask detects it will start producing data again, it should emit a
"watermark reactivate" message. Actual watermarks will as a result be produced
again from the data of that source subtask (or timestamp extractor/watermark
assigners that read from that source subtask).
- In the case that subtasks initially don't have partitions and get assigned
one afterwards, it should behave likewise the above description.
- Concerning rogue timestamp extractor/watermark assigners in the middle of a
topology that cause some instances to stop producing watermarks infinitely:
sticking to the nature that "event watermarks are derived from the data
itself", I think it's the user implementation's responsibility that this
shouldn't happen. As long as there are data flowing through the timestamp
extractor/watermark assigner instance, the implementation should guarantee that
watermarks make progress. In the case where no data is flowing through the
extractor/watermark assigner instance, this should be internally taken care of
by the "watermark idle" / "watermark reactivate" messages sent from the sources.
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
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15632701#comment-15632701
]
Aljoscha Krettek commented on FLINK-4576:
-
It's a good solution for the problem of idle sources (where we know that they
are idle because they don't have partitions assigned).
I don't think it solves the problem of having partitions assigned that don't
produce new elements for a while. I also don't think it solves the problem of
having a timestamp extractor/watermark assigner in the middle of a topology,
event right after the sources. Does it?
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15632615#comment-15632615
]
Stephan Ewen commented on FLINK-4576:
-
+1 for the "watermark idle" messages approach. That strikes me as a good idea.
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
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 i
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15631789#comment-15631789
]
Tzu-Li (Gordon) Tai commented on FLINK-4576:
Hi [~aljoscha],
I think there's another problem for intermediate operators to work with the low
watermark service.
Say there are 3 instances 0, 1, 2 of an operator, and instance 1 is not
advancing it's watermark because one of it's input channels is not sending
watermarks. With the watermark service, we can combine the actual low watermark
across all instances for this operator, and let all instances get notified of
the global watermark. However, when instance 1 get's notified of this, we can't
really decide whether we really want to simply advance the watermark of
instance 1 - we don't actually know whether it's watermarks simply hasn't
arrived yet for some input channel, or there really is an input channel that's
not sending watermarks.
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15631405#comment-15631405
]
Tzu-Li (Gordon) Tai commented on FLINK-4576:
Ah, right. Got lost on this part mid-discussion.
However, I have the feeling that letting all operators interact with this
service (as pointed out previously, basically all stream operators need to work
with the service) there's a bit of overkill in trying to solve what initially
brought out the discussion of the low watermark service in the first place:
some source subtasks like FlinkKafkaConsumer may initially be idle with no
partitions to read from, in which we would want to emit the global low
watermark across subtasks for them, instead of the max value watermark. So, in
the beginning we were trying to aim for a source-only solution.
But honestly I currently don't have better ideas on how to achieve that,
because we allow users to assign watermarks at sources and in middle of
topologies.
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15629137#comment-15629137
]
Aljoscha Krettek commented on FLINK-4576:
-
Hi,
I think the watermark needs to be collected across the parallel instances of an
operator. For example, if you have a source, you would collect the current
watermark from all instances of that source, combine it and then send out the
minimum watermark to all those source instances.
If there are several operators in the topology that would need the watermark
notification functionality then this process has to be done for each operator
separately, i.e. for all parallel instances of one operator.
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15617827#comment-15617827
]
Tzu-Li (Gordon) Tai commented on FLINK-4576:
Hi [~aljoscha],
Thanks a lot for the pointers, they answer some of my doubts. I was thinking
hard about this, as well as finishing off some other tasks, hence the late
reply.
So, if I understand you correctly, the idea is that the current global
watermark has to be determined from watermarks collected from the whole
topology, and not only the operators within the chain of the source task,
correct?
Concerning new interfaces:
I think {{currentWatermark()}} is only needed for all operators in general
which users shouldn't need to implement and should only be internally used, but
{{notifyCurrentGlobalWatermark()}} is needed by the actual source operator user
implementations (so that source UDFs can determine whether or not they need to
emit the global watermark). So I am leaning towards separate interfaces for
this (perhaps an internal {{WatermarkOperator}} interface and a public
{{GlobalWatermarkListener}} interface), or am I missing anything in the picture?
Giving a second go at this task next week!
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15572142#comment-15572142
]
Aljoscha Krettek commented on FLINK-4576:
-
Hi,
first off, sorry for the late response, I was on vacation until this Tuesday.
I think you analysed the problems correctly:
1. Should both sources and intermediate operators be able to interact with
this new service?
2. How to get the watermark from operators and how to give it to operators?
I think you already have the answer to 1.: yes. Because we allow users to
assign watermarks in the middle of a pipeline and even provide "sources"
ourselves that do this (Kinesis, ContinuousFileSource).
This leads naturally to an answer for 2.: I think we need to introduce an
interface like this:
{code}
interface WatermarkOperator {
Watermark currentWatermark();
void notifyOfCurrentGlobalWatermark(Watermark watermark);
}
{code}
sources would implement this interface and any operators that we have
in-between would also implement it. This then means, of course, that basically
all stream tasks have to be able to handle watermarks and always ask all the
operators in the chain and update them if they implement the interface.
[~StephanEwen] do you maybe also have some thoughts on this?
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15535824#comment-15535824
]
Tzu-Li (Gordon) Tai commented on FLINK-4576:
Hi [~aljoscha],
I'm struggling a bit on implementation choice of recording and accessing the
last emitted watermark of each source operator.
When a {{SourceStreamTask}} receives a request to return its last emitted
watermark to JobManager, it accesses the last emitted watermark of the head
operator (a {{StreamSource}} operator) by calling
{{headOperator.getLastEmittedWatermark()}}.
I'm currently jumping around two possible ways to achieve this:
1. Record the last emitted watermark within {{SourceFunction.SourceContext}} s,
and expose it through a getter method.
{{headOperator.getLastEmittedWatermark()}} simply uses this getter method on
the source context.
2. Extend the `o.a.f.streaming.api.operators.Output` interface to have a new
{{getLastEmittedWatermark}} method, and let
{{headOperator.getLastEmittedWatermark()}} call that instead.
I'm not sure which would be a better approach?
Another problem I'd like to confirm is that the low watermark service should
only be relevant to watermarks emitted from {{StreamSource}} operators, and not
the watermarks emitted at the end of the whole operator chain of a
{{SourceStreamTask}}, correct?
I'm thinking about this because I'm a bit unsure with how user-supplied
watermark assigners should work with the JM low watermark service.
I've completed the communication between JM / TM / tasks, so I'm down to this
last bit and can open the PR afterwards. Any feedback will be helpful!
> 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15489818#comment-15489818
]
Tzu-Li (Gordon) Tai commented on FLINK-4576:
Hi [~aljoscha], thanks for the help!
I've actually only just started digging in two days ago, but yes the
implementation description I outlined above is basically following how the
{{CheckpointCoordiantor}} is implemented.
I hope to get a PR out for this by the end of the week, maybe initially without
tests so that I can get some early feedback from you :)
One thing I'm not entirely sure of is this part:
"The messages will only be relevant to tasks that implement an internal
{{LowWatermarkCooperatingTask}} interface. For now, only {{SourceStreamTask}}
should implement {{LowWatermarkCooperatingTask}}."
It is sort of following the way checkpoint messages are only relevant to
{{StatefulTask}}s, but somehow I have a feeling there might be a better layout
for this. 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)
[jira] [Commented] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources
[
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15489792#comment-15489792
]
Aljoscha Krettek commented on FLINK-4576:
-
Hi [~tzulitai],
how far along are you with the implementation? I recently worked on the
{{CheckpointCoordiantor}} so I know pretty well how the communication between
all the different parts works and the watermark service would work pretty
similar to that, I imagine.
Just let me know if you would like to bounce around some ideas or if you need
some input. :-)
> 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)
