Re: [DISCUSS] Support Local Aggregation in Flink

[vino yang]

Hi Dian,

I still think your implementation is similar to the window operator, you
mentioned the scalable trigger mechanism, the window API also can customize
trigger.

Moreover, IMO, the design should guarantee a deterministic semantics, I
think based on memory availability is a non-deterministic design.

In addition, if the implementation depends on the timing of checkpoint, I
do not think it is reasonable, we should avoid affecting the checkpoint's
progress.

Best,
Vino

Dian Fu  于2019年6月5日周三 下午1:55写道：

> Hi Vino,
>
> Thanks a lot for your reply.
>
> > 1) When, Why and How to judge the memory is exhausted?
>
> My point here is that the local aggregate operator can buffer the inputs
> in memory and send out the results AT ANY TIME. i.e. element count or the
> time interval reached a pre-configured value, the memory usage of buffered
> elements reached a configured valued (suppose we can estimate the object
> size efficiently), or even when checkpoint is triggered.
>
> >
> > 2) If the local aggregate operator rarely needs to operate the state,
> what
> > do you think about fault tolerance?
>
> AbstractStreamOperator provides a method `prepareSnapshotPreBarrier` which
> can be used here to send out the results to the downstream when checkpoint
> is triggered. Then fault tolerance can work well.
>
> Even if there is no such a method available, we can still store the
> buffered elements or pre-aggregate results to state when checkpoint is
> triggered. The state access will be much less compared with window operator
> as only the elements not sent out when checkpoint occur have to be written
> to state. Suppose the checkpoint interval is 3 minutes and the trigger
> interval is 10 seconds, then only about less than "10/180" elements will be
> written to state.
>
>
> Thanks,
> Dian
>
>
> > 在 2019年6月5日，上午11:43，Biao Liu  写道：
> >
> > Hi Vino,
> >
> > +1 for this feature. It's useful for data skew. And it could also reduce
> > shuffled datum.
> >
> > I have some concerns about the API part. From my side, this feature
> should
> > be more like an improvement. I'm afraid the proposal is an overkill about
> > the API part. Many other systems support pre-aggregation as an
> optimization
> > of global aggregation. The optimization might be used automatically or
> > manually but with a simple API. The proposal introduces a series of
> > flexible local aggregation APIs. They could be independent with global
> > aggregation. It doesn't look like an improvement but introduces a lot of
> > features. I'm not sure if there is a bigger picture later. As for now the
> > API part looks a little heavy for me.
> >
> >
> > vino yang  于2019年6月5日周三 上午10:38写道：
> >
> >> Hi Litree,
> >>
> >> From an implementation level, the localKeyBy API returns a general
> >> KeyedStream, you can call all the APIs which KeyedStream provides, we
> did
> >> not restrict its usage, although we can do this (for example returns a
> new
> >> stream object named LocalKeyedStream).
> >>
> >> However, to achieve the goal of local aggregation, it only makes sense
> to
> >> call the window API.
> >>
> >> Best,
> >> Vino
> >>
> >> litree  于2019年6月4日周二 下午10:41写道：
> >>
> >>> Hi Vino，
> >>>
> >>>
> >>> I have read your design，something I want to know is the usage of these
> >> new
> >>> APIs.It looks like when I use localByKey,i must then use a window
> >> operator
> >>> to return a datastream，and then use keyby and another window operator
> to
> >>> get the final result?
> >>>
> >>>
> >>> thanks,
> >>> Litree
> >>>
> >>>
> >>> On 06/04/2019 17:22, vino yang wrote:
> >>> Hi Dian,
> >>>
> >>> Thanks for your reply.
> >>>
> >>> I know what you mean. However, if you think deeply, you will find your
> >>> implementation need to provide an operator which looks like a window
> >>> operator. You need to use state and receive aggregation function and
> >>> specify the trigger time. It looks like a lightweight window operator.
> >>> Right?
> >>>
> >>> We try to reuse Flink provided functions and reduce complexity. IMO, It
> >> is
> >>> more user-friendly because users are familiar with the window API.
> >>>
> >>> Best,
> >>> Vino
> >>>
> >>>
> >>> Dian Fu  于2019年6月4日周二 下午4:19写道：
> >>>
>  Hi Vino,
> 
>  Thanks a lot for starting this discussion. +1 to this feature as I
> >> think
>  it will be very useful.
> 
>  Regarding to using window to buffer the input elements, personally I
> >>> don't
>  think it's a good solution for the following reasons:
>  1) As we know that WindowOperator will store the accumulated results
> in
>  states, this is not necessary for Local Aggregate operator.
>  2) For WindowOperator, each input element will be accumulated to
> >> states.
>  This is also not necessary for Local Aggregate operator and storing
> the
>  input elements in memory is enough.
> 
>  Thanks,
>  Dian
> 
> > 在 2019年6月4日，上午10:03，vino yang  写道：
> >
> > Hi Ken,
> >
> > Thanks for your reply.
> 

Re: [DISCUSS] Support Local Aggregation in Flink

[vino yang]

Hi Aljoscha,

What do you think about this feature and design document?

Best,
Vino

vino yang  于2019年6月5日周三 下午4:18写道：

> Hi Dian,
>
> I still think your implementation is similar to the window operator, you
> mentioned the scalable trigger mechanism, the window API also can customize
> trigger.
>
> Moreover, IMO, the design should guarantee a deterministic semantics, I
> think based on memory availability is a non-deterministic design.
>
> In addition, if the implementation depends on the timing of checkpoint, I
> do not think it is reasonable, we should avoid affecting the checkpoint's
> progress.
>
> Best,
> Vino
>
> Dian Fu  于2019年6月5日周三 下午1:55写道：
>
>> Hi Vino,
>>
>> Thanks a lot for your reply.
>>
>> > 1) When, Why and How to judge the memory is exhausted?
>>
>> My point here is that the local aggregate operator can buffer the inputs
>> in memory and send out the results AT ANY TIME. i.e. element count or the
>> time interval reached a pre-configured value, the memory usage of buffered
>> elements reached a configured valued (suppose we can estimate the object
>> size efficiently), or even when checkpoint is triggered.
>>
>> >
>> > 2) If the local aggregate operator rarely needs to operate the state,
>> what
>> > do you think about fault tolerance?
>>
>> AbstractStreamOperator provides a method `prepareSnapshotPreBarrier`
>> which can be used here to send out the results to the downstream when
>> checkpoint is triggered. Then fault tolerance can work well.
>>
>> Even if there is no such a method available, we can still store the
>> buffered elements or pre-aggregate results to state when checkpoint is
>> triggered. The state access will be much less compared with window operator
>> as only the elements not sent out when checkpoint occur have to be written
>> to state. Suppose the checkpoint interval is 3 minutes and the trigger
>> interval is 10 seconds, then only about less than "10/180" elements will be
>> written to state.
>>
>>
>> Thanks,
>> Dian
>>
>>
>> > 在 2019年6月5日，上午11:43，Biao Liu  写道：
>> >
>> > Hi Vino,
>> >
>> > +1 for this feature. It's useful for data skew. And it could also reduce
>> > shuffled datum.
>> >
>> > I have some concerns about the API part. From my side, this feature
>> should
>> > be more like an improvement. I'm afraid the proposal is an overkill
>> about
>> > the API part. Many other systems support pre-aggregation as an
>> optimization
>> > of global aggregation. The optimization might be used automatically or
>> > manually but with a simple API. The proposal introduces a series of
>> > flexible local aggregation APIs. They could be independent with global
>> > aggregation. It doesn't look like an improvement but introduces a lot of
>> > features. I'm not sure if there is a bigger picture later. As for now
>> the
>> > API part looks a little heavy for me.
>> >
>> >
>> > vino yang  于2019年6月5日周三 上午10:38写道：
>> >
>> >> Hi Litree,
>> >>
>> >> From an implementation level, the localKeyBy API returns a general
>> >> KeyedStream, you can call all the APIs which KeyedStream provides, we
>> did
>> >> not restrict its usage, although we can do this (for example returns a
>> new
>> >> stream object named LocalKeyedStream).
>> >>
>> >> However, to achieve the goal of local aggregation, it only makes sense
>> to
>> >> call the window API.
>> >>
>> >> Best,
>> >> Vino
>> >>
>> >> litree  于2019年6月4日周二 下午10:41写道：
>> >>
>> >>> Hi Vino，
>> >>>
>> >>>
>> >>> I have read your design，something I want to know is the usage of these
>> >> new
>> >>> APIs.It looks like when I use localByKey,i must then use a window
>> >> operator
>> >>> to return a datastream，and then use keyby and another window operator
>> to
>> >>> get the final result?
>> >>>
>> >>>
>> >>> thanks,
>> >>> Litree
>> >>>
>> >>>
>> >>> On 06/04/2019 17:22, vino yang wrote:
>> >>> Hi Dian,
>> >>>
>> >>> Thanks for your reply.
>> >>>
>> >>> I know what you mean. However, if you think deeply, you will find your
>> >>> implementation need to provide an operator which looks like a window
>> >>> operator. You need to use state and receive aggregation function and
>> >>> specify the trigger time. It looks like a lightweight window operator.
>> >>> Right?
>> >>>
>> >>> We try to reuse Flink provided functions and reduce complexity. IMO,
>> It
>> >> is
>> >>> more user-friendly because users are familiar with the window API.
>> >>>
>> >>> Best,
>> >>> Vino
>> >>>
>> >>>
>> >>> Dian Fu  于2019年6月4日周二 下午4:19写道：
>> >>>
>>  Hi Vino,
>> 
>>  Thanks a lot for starting this discussion. +1 to this feature as I
>> >> think
>>  it will be very useful.
>> 
>>  Regarding to using window to buffer the input elements, personally I
>> >>> don't
>>  think it's a good solution for the following reasons:
>>  1) As we know that WindowOperator will store the accumulated results
>> in
>>  states, this is not necessary for Local Aggregate operator.
>>  2) For WindowOperator, each input element will be accumulated to
>> >> s

[jira] [Created] (FLINK-12736) ResourceManager may release TM with allocated slots

[Chesnay Schepler (JIRA)]

Chesnay Schepler created FLINK-12736:


 Summary: ResourceManager may release TM with allocated slots
 Key: FLINK-12736
 URL: https://issues.apache.org/jira/browse/FLINK-12736
 Project: Flink
  Issue Type: Bug
  Components: Runtime / Coordination
Affects Versions: 1.9.0
Reporter: Chesnay Schepler
 Fix For: 1.9.0


The {{ResourceManager}} looks out for TaskManagers that have not had any slots 
allocated on them for a while, as these could be released to safe resources. If 
such a TM is found the RM checks via an RPC call whether the TM still holds any 
partitions. If no partition is held then the TM is released.

However, in the RPC callback no check is made whether the TM is actually 
_still_ idle. In the meantime a slot could've been allocated on the TM.





--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Re: [DISCUSS] FLIP-41: Unify Keyed State Snapshot Binary Format for Savepoints

[Tzu-Li (Gordon) Tai]

I want to quickly bump this discussion to gather more consensus from others
on the FLIP, and see if we want to aim this for the upcoming 1.9.0 release.
The proposal touches binary formats of savepoints, which is a major part of
Flink's public user interface, so having explicit approval from other
members of the community would be nice here.

Cheers,
Gordon

On Wed, May 29, 2019 at 11:45 AM Tzu-Li (Gordon) Tai 
wrote:

> I also should point out something that I forgot to mention in the initial
> post:
> Stefan has helped a lot in understanding the current status of state
> backends and also participated a lot in design choices for the FLIP :)
>
> On Wed, May 29, 2019 at 5:02 AM Tzu-Li (Gordon) Tai 
> wrote:
>
>> Hi Flink devs,
>>
>> Congxian, Kostas, and I have recently been discussing to unify the binary
>> formats for keyed state in savepoints, which would allow for more
>> operational flexibility such as swapping state backends across restores.
>>
>> As part of this FLIP, another main proposal is to start allowing
>> checkpoints and savepoints to have different formats. Savepoint formats
>> should in the future be designed with interoperability in mind and
>> reasonable snapshot / restore overhead is tolerable, while checkpoints are
>> allowed to be backend specific for more efficient snapshots and restores.
>> From recent proposals in the state backends such as disk-spilling heap
>> backend [1], this flexibility seems to be reasonable.
>>
>> The main user-facing API this would affect is of course, the binary
>> formats of savepoints, as well as the fact that we will no longer be
>> guaranteeing functional parity between savepoints and full checkpoints in
>> the future (w.r.t. operational features related to upgrading applications;
>> so far they have equal functionality).
>>
>> Therefore, we would like to collect feedback on the proposal before
>> continuing efforts.
>>
>> This is the FLIP:
>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-41%3A+Unify+Keyed+State+Snapshot+Binary+Format+for+Savepoints
>> .
>>
>> I'm happy to discuss details and looking forward to any feedback.
>>
>> Cheers,
>> Gordon
>>
>> [1]
>> http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-Proposal-to-support-disk-spilling-in-HeapKeyedStateBackend-td29109.html
>>
>

Re: [DISCUSS] Releasing Flink 1.8.1

[jincheng sun]

I am here to quickly update the progress of the issue that needs to be
tracked(going well):

[Blocker]
- FLINK-12296  [done]
- FLINK-11987  [done]
- FLINK-12297   Being
reviewed by @Aljoscha Krettek !
- FLINK-11107   Being
reviewed by @Tzu-Li (Gordon) Tai 

[Critical]
- FLINK-10455   Will
open the PR soon, great job @Jiangjie Qin
 ！
- FLINK-11059   Being
reviewed by @Till Rohrmann !

[Nice to have]
- FLINK-12544  [done]

The detail can be found here:
https://docs.google.com/document/d/1858C7HdyDPIxxm2Rvnu4bYahq0Tr9NaY1mj7BzQi_0w/edit?usp=sharing

Great thanks to all of you for the help(fix or review) in promoting the
1.8.1 release. Thank you!!!

BTW: That's great if we can fix all of those issues, and prepare the first
RC of release 1.8.1 next week. :)

Best,
Jincheng

jincheng sun  于2019年6月3日周一 下午1:22写道：

> I am here to quickly update the progress of the issue that needs to be
> tracked:
>
> [Blocker]
> - FLINK-12296  [done]
> - FLINK-11987  [done]
> - FLINK-12297   @Aljoscha
> Krettek  will review the PR!
> - FLINK-11107  @Tzu-Li
> (Gordon) Tai  will help to review! (This is a new
> captur）
>
> [Critical]
> - FLINK-10455   @Jiangjie
> Qin
>  will
> help to take the ticket!
> - FLINK-11059   Being
> reviewed by @Till Rohrmann !
>
> [Nice to have]
> - FLINK-12544  Being
> reviewed by @Piotr Nowojski 
>
> The detail can be found here:
>
> https://docs.google.com/document/d/1858C7HdyDPIxxm2Rvnu4bYahq0Tr9NaY1mj7BzQi_0w/edit?usp=sharing
>
> Great thanks to all of you for the help(fix or review) in promoting the
> 1.8.1 release. Thank you!!!
>
> BTW: That's great if we can fix all of those issues, and prepare the first
> RC of release 1.8.1 next week. :)
>
> Best,
> Jincheng
>
>
> Jark Wu  于2019年5月30日周四 下午9:49写道：
>
>> Hi Jingcheng,
>>
>> Thanks for coordinating the work to release 1.8.1.
>>
>> +1 for 1.8.1
>>
>> On Wed, 29 May 2019 at 19:48, Hequn Cheng  wrote:
>>
>> > Hi Jincheng,
>> >
>> > Thanks for putting these together with a nice document.
>> > +1 to release 1.8.1. I think it would be nice if we can have a new
>> release
>> > with so many fixes.
>> >
>> > Best, Hequn
>> >
>> > On Wed, May 29, 2019 at 5:25 PM jincheng sun 
>> > wrote:
>> >
>> > > Hi all,
>> > > Thank you for your support of the release of 1.8.1.
>> > >
>> > > @Till Rohrmann   Thank you very much for your
>> help
>> > > review FLINK-11059!
>> > > @Zhijiang Thank you feedback the very important bug fix. I'll add it
>> to
>> > the
>> > > trace list!
>> > > @Tzu-Li (Gordon) Tai  Great thanks for your can
>> > > kindly
>> > > help for the final stage for the flink 1.8.1 release!
>> > > @Congxian Thanks for your double check the FLINK-12296 and update the
>> > > status!
>> > >
>> > > I am here to update the progress of the issue that needs to be
>> tracked:
>> > >
>> > > [Blocker]
>> > > - FLINK-12296 
>> [done]
>> > > - FLINK-11987 
>> > > KafkSolution needs discussion, great thanks @Yu Li 
>> > > and @Becketqin join
>> > > in the discussion!
>> > > - FLINK-12297 
>> > > @Aljoscha
>> > > Krettek  will review the PR  next week! Great
>> > thanks
>> > > to @Aljoscha Krettek !
>> > >
>> > > [Critical]
>> > > - FLINK-10455  I
>> > left a
>> > > comment on JIRA and need further discussion. (May need
>> > > - FLINK-11059 
>> Great
>> > > thanks to @Till Rohrmann  help to review.
>> > >
>> > > [Nice to have]
>> > > - FLINK-12544 
>> Great
>> > > thanks for the PR @Zhijiang, and need someone to volunteer review the
>> PR!
>> > >
>> > > Furthermore, in order to better track the progress and discussion of
>> > > issues, I have compiled a simple google doc, welcome any feedback and
>> > > comments.
>> > >
>> > >
>> >
>> https://docs.google.com/document/d/1858C7HdyDPIxxm2Rvnu4bYahq0Tr9NaY1mj7BzQi_0w/edit?usp=sharing
>> > >
>> > > Best,
>> > > Jincheng
>> > >
>> > >
>> > > Congxian Qiu  于2019年5月29日周三 上午10:47写道：
>> 

[jira] [Created] (FLINK-12737) Loosen Table dependencies

[Dawid Wysakowicz (JIRA)]

Dawid Wysakowicz created FLINK-12737:


 Summary: Loosen Table dependencies
 Key: FLINK-12737
 URL: https://issues.apache.org/jira/browse/FLINK-12737
 Project: Flink
  Issue Type: Sub-task
  Components: Table SQL / API, Table SQL / Legacy Planner
Reporter: Dawid Wysakowicz
Assignee: Dawid Wysakowicz
 Fix For: 1.9.0


The aim of this task is to remove dependencies on actual implementation of 
{{TableEnvironment}}. This includes:

* drop generating unique attribute generation (it sufficient to index 
aggregates within a single operation)
* make the transformation from {{Table}} on the caller side rather than on the 
callee (a.k.a remove getRelNode)
* Add {{insertInto}} method to {{TableEnvironment}}

Additionally move the TemporalTableFunctionImpl to api-java module.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12738) Remove abstract getPageSize from InputGate

[zhijiang (JIRA)]

zhijiang created FLINK-12738:


 Summary: Remove abstract getPageSize from InputGate
 Key: FLINK-12738
 URL: https://issues.apache.org/jira/browse/FLINK-12738
 Project: Flink
  Issue Type: Sub-task
  Components: Runtime / Network
Reporter: zhijiang
Assignee: zhijiang


Currently {{InputGate#getPageSize}} is only used for constructing 
{{BarrierBuffer}}. In order to make abstract InputGate simple and clean, we 
should remove unnecessary abstract methods from it.

Considering the page size could be parsed directly from configuration which 
could also visible while constructing {{BarrierBuffer}}, so it is reasonable to 
do so.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12739) Flink does not build with scala 2.12

[Chesnay Schepler (JIRA)]

Chesnay Schepler created FLINK-12739:


 Summary: Flink does not build with scala 2.12
 Key: FLINK-12739
 URL: https://issues.apache.org/jira/browse/FLINK-12739
 Project: Flink
  Issue Type: Bug
  Components: Build System, Table SQL / API
Affects Versions: 1.9.0
Reporter: Chesnay Schepler
 Fix For: 1.9.0


https://travis-ci.org/apache/flink/jobs/541175138

Bit of a weird one:

{code}
12:02:31.766 [INFO] flink-sql-parser ... 
FAILURE [  0.005 s]
12:02:31.766 [INFO] flink-table-planner-blink .. SKIPPED
12:02:31.766 [INFO] flink-contrib .. SKIPPED
12:02:31.766 [INFO] flink-connector-wikiedits .. SKIPPED
12:02:31.766 [INFO] flink-yarn-tests ... SKIPPED
12:02:31.766 [INFO] flink-fs-tests . SKIPPED
12:02:31.766 [INFO] flink-docs . SKIPPED
12:02:31.766 [INFO] flink-ml-parent  SKIPPED
12:02:31.766 [INFO] flink-ml-api ... SKIPPED
12:02:31.766 [INFO] 

12:02:31.766 [INFO] BUILD FAILURE
12:02:31.766 [INFO] 

12:02:31.766 [INFO] Total time: 25:06 min
12:02:31.766 [INFO] Finished at: 2019-06-04T12:02:31+00:00
12:02:34.382 [INFO] Final Memory: 460M/1232M
12:02:34.382 [INFO] 

12:02:34.384 [ERROR] Could not find goal 'regex-property' in plugin 
org.codehaus.mojo:build-helper-maven-plugin:1.5 among available goals 
reserve-network-port, parse-version, maven-version, add-test-resource, 
add-test-source, add-resource, attach-artifact, add-source, 
remove-project-artifact, help -> [Help 1]
{code}



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12740) SpillableSubpartitionTest deadlocks on Travis

[Chesnay Schepler (JIRA)]

Chesnay Schepler created FLINK-12740:


 Summary: SpillableSubpartitionTest deadlocks on Travis
 Key: FLINK-12740
 URL: https://issues.apache.org/jira/browse/FLINK-12740
 Project: Flink
  Issue Type: Bug
  Components: Runtime / Network
Affects Versions: 1.8.1
Reporter: Chesnay Schepler
 Fix For: 1.8.1


https://travis-ci.org/apache/flink/jobs/541225542



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Re: [DISCUSS] FLIP-33: Standardize connector metrics

[Piotr Nowojski]

Hi Becket,

Thanks for the answer :)

> By time-based metric, I meant the portion of time spent on producing the
> record to downstream. For example, a source connector can report that it's
> spending 80% of time to emit record to downstream processing pipeline. In
> another case, a sink connector may report that its spending 30% of time
> producing the records to the external system.
> 
> This is in some sense equivalent to the buffer usage metric:

>   - 80% of time spent on emitting records to downstream ---> downstream
> node is bottleneck ---> output buffer is probably full.
>   - 30% of time spent on emitting records to downstream ---> downstream
> node is not bottleneck ---> output buffer is probably not full.

If by “time spent on emitting records to downstream” you understand “waiting on 
back pressure”, then I see your point. And I agree that some kind of ratio/time 
based metric gives you more information. However under “time spent on emitting 
records to downstream” might be hidden the following (extreme) situation:

1. Job is barely able to handle influx of records, there is 99% CPU/resource 
usage in the cluster, but nobody is bottlenecked/backpressured, all output 
buffers are empty, everybody is waiting in 1% of it’s time for more records to 
process.
2. 80% time can still be spent on "down stream operators”, because they are the 
CPU intensive operations, but this doesn’t mean that increasing the parallelism 
down the stream will help with anything there. To the contrary, increasing 
parallelism of the source operator might help to increase resource utilisation 
up to 100%.

However, this “time based/ratio” approach can be extended to in/output buffer 
usage. Besides collecting an information that input/output buffer is 
full/empty, we can probe profile how often are buffer empty/full. If output 
buffer is full 1% of times, there is almost no back pressure. If it’s full 80% 
of times, there is some back pressure, if it’s full 99.9% of times, there is 
huge back pressure. 

Now for autoscaling you could compare the input & output buffers fill ratio:

1. Both are high, the source of bottleneck is down the stream
2. Output is low, input is high, this is the bottleneck and the higher the 
difference, the bigger source of bottleneck is this is operator/task
3. Output is high, input is low - there was some load spike that we are 
currently finishing to process



But long story short, we are probably diverging from the topic of this 
discussion, and we can discuss this at some later point.

For now, for sources:

as I wrote before, +1 for:
 - pending.bytes, Gauge
 - pending.messages, Gauge

When we will be developing/discussing SourceReader from FLIP-27 we might then 
add:

- in-memory.buffer.usage (0 - 100%)

Which will be estimated automatically by Flink while user will be able to 
override/provide better estimation.

Piotrek 

> On 5 Jun 2019, at 05:42, Becket Qin  wrote:
> 
> Hi Piotr,
> 
> Thanks for the explanation. Please see some clarifications below.
> 
> By time-based metric, I meant the portion of time spent on producing the
> record to downstream. For example, a source connector can report that it's
> spending 80% of time to emit record to downstream processing pipeline. In
> another case, a sink connector may report that its spending 30% of time
> producing the records to the external system.
> 
> This is in some sense equivalent to the buffer usage metric:
>   - 80% of time spent on emitting records to downstream ---> downstream
> node is bottleneck ---> output buffer is probably full.
>   - 30% of time spent on emitting records to downstream ---> downstream
> node is not bottleneck ---> output buffer is probably not full.
> 
> However, the time-based metric has a few advantages that the buffer usage
> metric may not have.
> 
> 1.  Buffer usage metric may not be applicable to all the connector
> implementations, while reporting time-based metric are always doable.
> Some source connectors may not have any input buffer, or they may use some
> third party library that does not expose the input buffer at all.
> Similarly, for sink connectors, the implementation may not have any output
> buffer, or the third party library does not expose such buffer.
> 
> 2. Although both type of metrics can detect bottleneck, time-based metrics
> can be used to generate a more informed action to remove the bottleneck.
> For example, when the downstream is bottleneck, the output buffer usage
> metric is likely to be 100%, and the input buffer usage might be 0%. That
> means we don't know what is the suitable parallelism to lift the
> bottleneck. The time-based metric, on the other hand, would give useful
> information, e.g. if 80% of time was spent on emitting records, we can
> roughly increase the downstream node parallelism by 4 times.
> 
> Admittedly, the time-based metrics are more expensive than buffer usage. So
> we may have to do some sampling to reduce the cost. But in general, using
> time-bas

[jira] [Created] (FLINK-12741) Update docs about Kafka fault tolerant guarantees

[Paul Lin (JIRA)]

Paul Lin created FLINK-12741:


 Summary: Update docs about Kafka fault tolerant guarantees
 Key: FLINK-12741
 URL: https://issues.apache.org/jira/browse/FLINK-12741
 Project: Flink
  Issue Type: Improvement
  Components: Documentation
Affects Versions: 1.9.0
Reporter: Paul Lin
Assignee: Paul Lin


Since Flink 1.4.0, we provide exactly-once semantic on Kafka 0.11+, but the 
document is still not updated.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12742) Add insert into partition grammar as hive dialect

[Danny Chan (JIRA)]

Danny Chan created FLINK-12742:
--

 Summary: Add insert into partition grammar as hive dialect
 Key: FLINK-12742
 URL: https://issues.apache.org/jira/browse/FLINK-12742
 Project: Flink
  Issue Type: Sub-task
  Components: Table SQL / API
Affects Versions: 1.9.0
Reporter: Danny Chan
Assignee: Danny Chan
 Fix For: 1.9.0






--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12743) Introduce unbounded streaming anti/semi join operator

[Jark Wu (JIRA)]

Jark Wu created FLINK-12743:
---

 Summary: Introduce unbounded streaming anti/semi join operator
 Key: FLINK-12743
 URL: https://issues.apache.org/jira/browse/FLINK-12743
 Project: Flink
  Issue Type: New Feature
  Components: Table SQL / Runtime
Reporter: Jark Wu
Assignee: Jark Wu


This operator is responsible for unbounded streaming semi/anti join, and will 
be optimized in following cases:

1. If the join keys (with equality condition) are also primary key, we will 
have a more efficient state layout
2. If the inputs have primary keys, but join keys are not primary key, we can 
also come up with an efficient state layout
3. Inputs don't have primary keys, this will go to default implementation



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12744) ML common parameters

[Xu Yang (JIRA)]

Xu Yang created FLINK-12744:
---

 Summary: ML common parameters
 Key: FLINK-12744
 URL: https://issues.apache.org/jira/browse/FLINK-12744
 Project: Flink
  Issue Type: Sub-task
Reporter: Xu Yang
Assignee: Xu Yang


We defined some common-used parameters for machine-learning algorithms.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12745) Sparse and

[Xu Yang (JIRA)]

Xu Yang created FLINK-12745:
---

 Summary: Sparse and 
 Key: FLINK-12745
 URL: https://issues.apache.org/jira/browse/FLINK-12745
 Project: Flink
  Issue Type: Sub-task
Reporter: Xu Yang






--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12746) Getting Started - Project Setup and DataStream Example Walkthrough

[Konstantin Knauf (JIRA)]

Konstantin Knauf created FLINK-12746:


 Summary: Getting Started - Project Setup and DataStream Example 
Walkthrough
 Key: FLINK-12746
 URL: https://issues.apache.org/jira/browse/FLINK-12746
 Project: Flink
  Issue Type: Sub-task
Reporter: Konstantin Knauf


The planned structure for the new Getting Started Guide is 

*  Flink Overview (~ two pages)
* Project Setup
* Quickstarts
** Example Walkthrough - Table API / SQL
** Example Walkthrough - DataStream API
* Docker Playgrounds
** Flink Cluster Playground
** Flink Interactive SQL Playground

In this ticket we should add "Project Setup" and "Quickstarts -> Example 
Walkthrough - DataStream API", which covers everything what we have today. This 
will replace the current "Tutorials" and "Examples" section, which can be 
removed as part of this ticket as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12747) Getting Started - Table API Example Walkthrough

[Konstantin Knauf (JIRA)]

Konstantin Knauf created FLINK-12747:


 Summary: Getting Started - Table API Example Walkthrough
 Key: FLINK-12747
 URL: https://issues.apache.org/jira/browse/FLINK-12747
 Project: Flink
  Issue Type: Sub-task
Reporter: Konstantin Knauf


The planned structure for the new Getting Started Guide is 

*  Flink Overview (~ two pages)
* Project Setup
* Quickstarts
** Example Walkthrough - Table API / SQL
** Example Walkthrough - DataStream API
* Docker Playgrounds
** Flink Cluster Playground
** Flink Interactive SQL Playground

This tickets adds the Example Walkthrough for the Table API, which should 
follow the same structure as the DataStream Example (FLINK-12746), which needs 
to be completed first.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12748) Getting Started - Flink Overview

[Konstantin Knauf (JIRA)]

Konstantin Knauf created FLINK-12748:


 Summary: Getting Started - Flink Overview
 Key: FLINK-12748
 URL: https://issues.apache.org/jira/browse/FLINK-12748
 Project: Flink
  Issue Type: Sub-task
Reporter: Konstantin Knauf


The planned structure for the new Getting Started Guide is 

*  Flink Overview (~ two pages)
* Project Setup
* Quickstarts
** Example Walkthrough - Table API / SQL
** Example Walkthrough - DataStream API
* Docker Playgrounds
** Flink Cluster Playground
** Flink Interactive SQL Playground

In this ticket we should add a 1-2 page introduction & overview of Apache Flink 
including among other things an overview of the API Stack (DataStream API & 
SQL/Table API), an introduction to *stateful* stream processing and Flink's 
role in an overall stream processing architecture.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12749) FLINK-12748 - Docker Playgrounds - Flink Cluster Playground

[Konstantin Knauf (JIRA)]

Konstantin Knauf created FLINK-12749:


 Summary: FLINK-12748 - Docker Playgrounds - Flink Cluster 
Playground
 Key: FLINK-12749
 URL: https://issues.apache.org/jira/browse/FLINK-12749
 Project: Flink
  Issue Type: Sub-task
Reporter: Konstantin Knauf


The planned structure for the new Getting Started Guide is

Flink Overview (~ two pages)
Project Setup
Quickstarts
Example Walkthrough - Table API / SQL
Example Walkthrough - DataStream API
Docker Playgrounds
Flink Cluster Playground
Flink Interactive SQL Playground

In this ticket we add the Flink Cluster Playground, a docker-compose based 
setup consisting of Apache Kafka and Apache Flink (Flink Session Cluster), 
including a step-by-step guide for some common commands (job submission, 
savepoints, etc).

*Some Open Questions:*
* Which Flink images to use? `library/flink` with dynamic properties would be 
the most maintainable, I think. It would be preferable, if we don't need to 
host any custom images for this, but can rely on the existing plain Flink 
images.
* Which Flink jobs to use? An updated version 
{{org.apache.flink.streaming.examples.statemachine.StateMachineExample}} might 
be a good option as it can with or without Kafka and contains a data generator 
writing to Kafka already (see next questions).
* How to get data into Kafka? Maybe just provide a small bash script/one-liner 
to produce into Kafka topic or see question above.
* Which Kafka Images to use? https://hub.docker.com/r/wurstmeister/kafka/ seems 
to be well-maintained and is openly available.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12750) Gettting Started - Docker Playground - Interactive SQL Playground

[Konstantin Knauf (JIRA)]

Konstantin Knauf created FLINK-12750:


 Summary: Gettting Started - Docker Playground - Interactive SQL 
Playground
 Key: FLINK-12750
 URL: https://issues.apache.org/jira/browse/FLINK-12750
 Project: Flink
  Issue Type: Sub-task
Reporter: Konstantin Knauf


The planned structure for the new Getting Started Guide is

* Flink Overview (~ two pages)
* Project Setup
* Quickstarts
** Example Walkthrough - Table API / SQL
** Example Walkthrough - DataStream API
* Docker Playgrounds
** Flink Cluster Playground
** Flink Interactive SQL Playground

In this ticket we add the Flink Cluster Playground, a docker-compose based 
setup consisting of Apache Kafka and Apache Flink (Flink Session Cluster) and 
an SQL-Client. 

The general setup should be in line with FLINK-12749. 

**Open Questions**
* Where to host the SQL Client image? Can we somehow also use existing plain 
Flink images?



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12751) Create file based HA support

[Boris Lublinsky (JIRA)]

Boris Lublinsky created FLINK-12751:
---

 Summary: Create file based HA support
 Key: FLINK-12751
 URL: https://issues.apache.org/jira/browse/FLINK-12751
 Project: Flink
  Issue Type: Improvement
  Components: FileSystems
Affects Versions: 1.8.0, 1.9.0, 2.0.0
 Environment: Flink on k8 and Mini cluster
Reporter: Boris Lublinsky


In the current Flink implementation, HA support can be implemented either using 
Zookeeper or Custom Factory class.
Add HA implementation based on PVC. The idea behind this implementation
is as follows:
* Because implementation assumes a single instance of Job manager (Job manager 
selection and restarts are done by K8 Deployment of 1)
URL management is done using StandaloneHaServices implementation (in the case 
of cluster) and EmbeddedHaServices implementation (in the case of mini cluster)
* For management of the submitted Job Graphs, checkpoint counter and completed 
checkpoint an implementation is leveraging the following file system layout

 ha -> root of the HA data
 checkpointcounter -> checkpoint counter folder
  -> job id folder
  -> counter file
  -> another job id folder
 ...
 completedCheckpoint -> completed checkpoint folder
  -> job id folder
  -> checkpoint file
  -> checkpoint file
 ...
  -> another job id folder
 ...
 submittedJobGraph -> submitted graph folder
  -> job id folder
  -> graph file
  -> another job id folder
 ...




--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12752) Add Option to Pass Seed for JobID Hash for StandaloneJobClusterEntrypoint

[Konstantin Knauf (JIRA)]

Konstantin Knauf created FLINK-12752:


 Summary: Add Option to Pass Seed for JobID Hash for 
StandaloneJobClusterEntrypoint
 Key: FLINK-12752
 URL: https://issues.apache.org/jira/browse/FLINK-12752
 Project: Flink
  Issue Type: Sub-task
Reporter: Konstantin Knauf
Assignee: Konstantin Knauf


In case of a standalone job cluster setup, we would like to generate random a 
{{JobID}} for every job, but the {{JobID}} nevertheless needs to stay constant 
between JobManager process restarts. 

For this, I would like to add an additional command line options for the 
{{StandaloneJobClusterEntrypoint}} called {{jobId-seed}}. A manually specified 
jobId would still take precedence (not breaking current behavior): 

* --jobId
* --jobId-seed
* (default)

On Kubernetes, this new command line argument would need to be set to a 
property which is stable over Pod Restarts but changes for different K8s Jobs 
or K8s Deployments.





--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12753) Implement GenericInMemoryCatalog

[Bowen Li (JIRA)]

Bowen Li created FLINK-12753:


 Summary: Implement GenericInMemoryCatalog
 Key: FLINK-12753
 URL: https://issues.apache.org/jira/browse/FLINK-12753
 Project: Flink
  Issue Type: New Feature
  Components: Table SQL / API
Reporter: Bowen Li
Assignee: Bowen Li
 Fix For: 1.9.0


Introducing a generic in-memory catalog as session-based, non-persistent 
catalog for users to use by default.

Most parts of the catalog have been implemented along with introductions of 
catalog APIs. This umberlla jira is mainly for any smaller unimplemented 
features, refactors, and cleanups.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12754) alterTableColumnStatistics() in GenericInMemoryCatalog should ensure the table is non-partitioned

[Bowen Li (JIRA)]

Bowen Li created FLINK-12754:


 Summary: alterTableColumnStatistics() in GenericInMemoryCatalog 
should ensure the table is non-partitioned
 Key: FLINK-12754
 URL: https://issues.apache.org/jira/browse/FLINK-12754
 Project: Flink
  Issue Type: Sub-task
  Components: Table SQL / API
Reporter: Bowen Li
 Fix For: 1.9.0


Currently logic (below) doesn't check if the partition is partitioned or not.

{code:java}
@Override
public void alterTableStatistics(ObjectPath tablePath, 
CatalogTableStatistics tableStatistics, boolean ignoreIfNotExists)
throws TableNotExistException {
checkNotNull(tablePath);
checkNotNull(tableStatistics);

if (tableExists(tablePath)) {
tableStats.put(tablePath, tableStatistics.copy());
} else if (!ignoreIfNotExists) {
throw new TableNotExistException(getName(), tablePath);
}
}
{code}




--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12755) implement HiveCatalog

[Bowen Li (JIRA)]

Bowen Li created FLINK-12755:


 Summary: implement HiveCatalog
 Key: FLINK-12755
 URL: https://issues.apache.org/jira/browse/FLINK-12755
 Project: Flink
  Issue Type: New Feature
  Components: Connectors / Hive
Reporter: Bowen Li
Assignee: Bowen Li
 Fix For: 1.9.0


An umbrella JIRA for HiveCatalog.

Background: Previously we have GenericHiveMetastoreCatalog and HiveCatalog, and 
we decided to unify them into a single HiveCatalog with two code paths to deal 
with generic Flink metadata and Hive metadata.

This umbrella JIRA is for anything that is shared between the two code paths.

JIRA tickets related to one code path should be subtasks of FLINK-11479 or 
FLINK-11482



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12756) migrate HiveCatalog from TypeInformation-based old type system to DataType-based new type system

[Bowen Li (JIRA)]

Bowen Li created FLINK-12756:


 Summary: migrate HiveCatalog from TypeInformation-based old type 
system to DataType-based new type system
 Key: FLINK-12756
 URL: https://issues.apache.org/jira/browse/FLINK-12756
 Project: Flink
  Issue Type: Sub-task
  Components: Connectors / Hive
Reporter: Bowen Li
Assignee: Bowen Li
 Fix For: 1.9.0






--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12757) Improves the Python word_count example to use the descriptor API

[Dian Fu (JIRA)]

Dian Fu created FLINK-12757:
---

 Summary: Improves the Python word_count example to use the 
descriptor API
 Key: FLINK-12757
 URL: https://issues.apache.org/jira/browse/FLINK-12757
 Project: Flink
  Issue Type: Sub-task
  Components: API / Python
Reporter: Dian Fu
Assignee: Dian Fu


The aim of this ticket is to improve the word_count example:
1. Uses the from_element API to create a source table
2. Uses the descriptor API to register the sink



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12758) Add flink-ml-lib module

[Luo Gen (JIRA)]

Luo Gen created FLINK-12758:
---

 Summary: Add flink-ml-lib module
 Key: FLINK-12758
 URL: https://issues.apache.org/jira/browse/FLINK-12758
 Project: Flink
  Issue Type: Sub-task
  Components: Library / Machine Learning
Reporter: Luo Gen
Assignee: Luo Gen
 Fix For: 1.9.0


The Jira introduces a new module "flink-ml-lib" under flink-ml-parent.

The flink-ml-lib is planned in the roadmap in FLIP-39, as the code base of 
library implementations of FlinkML. This Jira only aims to create the module, 
and algorithms will be added in separate Jira in the future.

 For more details, please refer to [FLIP39 design 
doc|[https://docs.google.com/document/d/1StObo1DLp8iiy0rbukx8kwAJb0BwDZrQrMWub3DzsEo]]



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (FLINK-12759) FLInk localmode IN MINICluster can't drow DAG visualizer plan

[Wong (JIRA)]

Wong created FLINK-12759:


 Summary: FLInk localmode IN MINICluster can't drow DAG visualizer 
plan
 Key: FLINK-12759
 URL: https://issues.apache.org/jira/browse/FLINK-12759
 Project: Flink
  Issue Type: Bug
  Components: Runtime / Web Frontend
Affects Versions: 1.9.0
 Environment: WIN10 4core 16g
Reporter: Wong
 Fix For: 1.9.0
 Attachments: errorsinlocal.png, standalone.png

My plan: \{"nodes":[{"id":1,"type":"Source: fv_can","pact":"Data 
Source","contents":"Source: 
fv_can","parallelism":3},\{"id":2,"type":"CanCodec","pact":"Operator","contents":"CanCodec","parallelism":3,"predecessors":[{"id":1,"ship_strategy":"FORWARD","side":"second"}]},\{"id":3,"type":"Filter","pact":"Operator","contents":"Filter","parallelism":3,"predecessors":[{"id":2,"ship_strategy":"FORWARD","side":"second"}]},\{"id":5,"type":"Window(TumblingProcessingTimeWindows(1000),
 ProcessingTimeTrigger, 
Can2HbaseBatchFunction)","pact":"Operator","contents":"Window(TumblingProcessingTimeWindows(1000),
 ProcessingTimeTrigger, 
Can2HbaseBatchFunction)","parallelism":3,"predecessors":[{"id":3,"ship_strategy":"HASH","side":"second"}]},\{"id":6,"type":"Can2Redis","pact":"Operator","contents":"Can2Redis","parallelism":3,"predecessors":[{"id":3,"ship_strategy":"FORWARD","side":"second"}]},\{"id":7,"type":"Filter","pact":"Operator","contents":"Filter","parallelism":3,"predecessors":[{"id":2,"ship_strategy":"FORWARD","side":"second"}]},\{"id":8,"type":"RUDS","pact":"Operator","contents":"RUDS","parallelism":3,"predecessors":[{"id":7,"ship_strategy":"FORWARD","side":"second"}]}]}
 

 

 

it was well done on my standalone cluster v1.9-snapshot, but when i debug on 
localmode it has exceptions:   2019-06-06 11:43:01.283 
[flink-rest-server-netty-worker-thread-5] ERROR 
o.a.flink.runtime.rest.handler.job.JobVertexDetailsHandler - Could not create 
the handler request.
org.apache.flink.runtime.rest.handler.HandlerRequestException: Cannot resolve 
path parameter (vertexid) from value "details".
 at 
org.apache.flink.runtime.rest.handler.HandlerRequest.(HandlerRequest.java:71)
 ~[flink-runtime_2.12-1.9-SNAPSHOT.jar:1.9-SNAPSHOT]
 at 
org.apache.flink.runtime.rest.handler.AbstractHandler.respondAsLeader(AbstractHandler.java:143)
 ~[flink-runtime_2.12-1.9-SNAPSHOT.jar:1.9-SNAPSHOT]
 at 
org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.lambda$channelRead0$0(LeaderRetrievalHandler.java:81)
 [flink-runtime_2.12-1.9-SNAPSHOT.jar:1.9-SNAPSHOT]
 at java.util.Optional.ifPresent(Optional.java:159) ~[na:1.8.0_181]
 at org.apache.flink.util.OptionalConsumer.ifPresent(OptionalConsumer.java:44) 
~[flink-core-1.9-SNAPSHOT.jar:1.9-SNAPSHOT]
 at 
org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.channelRead0(LeaderRetrievalHandler.java:78)
 [flink-runtime_2.12-1.9-SNAPSHOT.jar:1.9-SNAPSHOT]
 at 
org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.channelRead0(LeaderRetrievalHandler.java:49)
 [flink-runtime_2.12-1.9-SNAPSHOT.jar:1.9-SNAPSHOT]
 at 
org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
 ~[flink-shaded-netty-4.1.32.Final-6.0.jar:na]
 at 
org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
 ~[flink-shaded-netty-4.1.32.Final-6.0.jar:na]
 at 
org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
 ~[flink-shaded-netty-4.1.32.Final-6.0.jar:na]
 at 
org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
 ~[flink-shaded-netty-4.1.32.Final-6.0.jar:na]
 at 
org.apache.flink.runtime.rest.handler.router.RouterHandler.routed(RouterHandler.java:110)
 ~[flink-runtime_2.12-1.9-SNAPSHOT.jar:1.9-SNAPSHOT]
 at 
org.apache.flink.runtime.rest.handler.router.RouterHandler.channelRead0(RouterHandler.java:89)
 ~[flink-runtime_2.12-1.9-SNAPSHOT.jar:1.9-SNAPSHOT]
 at 
org.apache.flink.runtime.rest.handler.router.RouterHandler.channelRead0(RouterHandler.java:54)
 ~[flink-runtime_2.12-1.9-SNAPSHOT.jar:1.9-SNAPSHOT]
 at 
org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
 ~[flink-shaded-netty-4.1.32.Final-6.0.jar:na]
 at 
org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
 ~[flink-shaded-netty-4.1.32.Final-6.0.jar:na]
 at 
org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
 ~[flink-shaded-netty-4.1.32.Final-6.0.jar:na]
 at 
org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
 ~[f

[jira] [Created] (FLINK-12760) Implement ExecutionGraph to InputsLocationsRetriever Adapter

[shuai.xu (JIRA)]

shuai.xu created FLINK-12760:


 Summary: Implement ExecutionGraph to InputsLocationsRetriever 
Adapter
 Key: FLINK-12760
 URL: https://issues.apache.org/jira/browse/FLINK-12760
 Project: Flink
  Issue Type: Sub-task
  Components: Runtime / Coordination
Reporter: shuai.xu
Assignee: shuai.xu
 Fix For: 1.9.0


Implement an [adapter|https://en.wikipedia.org/wiki/Adapter_pattern], which 
adapts the ExecutionGraph to the InputsLocationsRetriever interface.

*Acceptance criteria*
 * The adapter always reflects an up to date view of the ExecutionGraph state



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)