[GitHub] flink issue #4061: [FLINK-6841][table]Using TableSourceTable for both Stream...

[sunjincheng121]

Github user sunjincheng121 commented on the issue:

https://github.com/apache/flink/pull/4061
  
Hi @twalthr I think `DefinedProctimeAttribute` can be implemented by a 
batch table source. If we do not want a batch table source can implements 
`DefinedProctimeAttribute `, we should add move `DefinedProctimeAttribute` to 
`StreamTableSourceTable`. Otherwise, I feel that `StreamTableSourceTable` can 
be deleted. What do you think?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6841) using TableSourceTable for both Stream and Batch OR remove useless import

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6841:
---

Github user sunjincheng121 commented on the issue:

https://github.com/apache/flink/pull/4061
  
Hi @twalthr I think `DefinedProctimeAttribute` can be implemented by a 
batch table source. If we do not want a batch table source can implements 
`DefinedProctimeAttribute `, we should add move `DefinedProctimeAttribute` to 
`StreamTableSourceTable`. Otherwise, I feel that `StreamTableSourceTable` can 
be deleted. What do you think?


> using TableSourceTable for both Stream and Batch OR remove useless import
> -
>
> Key: FLINK-6841
> URL: https://issues.apache.org/jira/browse/FLINK-6841
> Project: Flink
>  Issue Type: Improvement
>  Components: Table API & SQL
>Affects Versions: 1.4.0
>Reporter: sunjincheng
>Assignee: sunjincheng
>
> 1. {{StreamTableSourceTable}} exist useless import of {{TableException}}
> 2. {{StreamTableSourceTable}} only override {{getRowType}} of  
> {{FlinkTable}}, I think we can override the method in {{TableSourceTable}}, 
> If so we can using {{TableSourceTable}} for both {{Stream}} and {{Batch}}.
> What do you think? [~fhueske] [~twalthr]



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6865) Expand checkstyle docs to include import in intellij

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6865:
---

Github user greghogan commented on the issue:

https://github.com/apache/flink/pull/4086
  
If the long-term plan is to have one checkstyle then we should keep the 
current plain name. We may be at the point where we can rename the checkstyles 
and update the in-progress modules to use a `relaxed-checkstyle.xml`.

I have not seen IntelliJ actually import settings from a checkstyle 
configuration. Not sure if this is user error or something is broken.


> Expand checkstyle docs to include import in intellij
> 
>
> Key: FLINK-6865
> URL: https://issues.apache.org/jira/browse/FLINK-6865
> Project: Flink
>  Issue Type: Improvement
>  Components: Checkstyle
>Affects Versions: 1.4.0
>Reporter: Chesnay Schepler
>Assignee: Chesnay Schepler
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink issue #4086: [FLINK-6865] Update checkstyle documentation

[greghogan]

Github user greghogan commented on the issue:

https://github.com/apache/flink/pull/4086
  
If the long-term plan is to have one checkstyle then we should keep the 
current plain name. We may be at the point where we can rename the checkstyles 
and update the in-progress modules to use a `relaxed-checkstyle.xml`.

I have not seen IntelliJ actually import settings from a checkstyle 
configuration. Not sure if this is user error or something is broken.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #4059: [FLINK-6830] Add StatefulJobSavepointFrom13Migrati...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/4059


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #4052: [FLINK-6808] Implement compatibility methods for C...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/4052


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6772) Incorrect ordering of matched state events in Flink CEP

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6772:
---

Github user kl0u commented on the issue:

https://github.com/apache/flink/pull/4084
  
Thanks @dawidwys ! I will let travis have another go on the rebased version 
and then merge.


> Incorrect ordering of matched state events in Flink CEP
> ---
>
> Key: FLINK-6772
> URL: https://issues.apache.org/jira/browse/FLINK-6772
> Project: Flink
>  Issue Type: Bug
>  Components: CEP
>Affects Versions: 1.3.0
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Kostas Kloudas
>  Labels: flink-rel-1.3.1-blockers
>
> I've stumbled across an unexepected ordering of the matched state events. 
> Pattern:
> {code}
> Pattern pattern = Pattern
> .begin("start")
> .where(new IterativeCondition() {
> @Override
> public boolean filter(String s, Context context) throws 
> Exception {
> return s.startsWith("a-");
> }
> }).times(4).allowCombinations()
> .followedByAny("end")
> .where(new IterativeCondition() {
> public boolean filter(String s, Context context) throws 
> Exception {
> return s.startsWith("b-");
> }
> }).times(3).consecutive();
> {code}
> Input event sequence:
> a-1, a-2, a-3, a-4, b-1, b-2, b-3
> On b-3 a matched pattern would be triggered.
> Now, in the {{Map}} map passed via {{select}} in 
> {{PatternSelectFunction}}, the list for the "end" state is:
> b-3, b-1, b-2.
> Based on the timestamp of the events (simply using processing time), the 
> correct order should be b-1, b-2, b-3.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6853) Migrating from Flink 1.1 fails for FlinkCEP

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6853:
---

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/4079


> Migrating from Flink 1.1 fails for FlinkCEP
> ---
>
> Key: FLINK-6853
> URL: https://issues.apache.org/jira/browse/FLINK-6853
> Project: Flink
>  Issue Type: Bug
>  Components: CEP
>Affects Versions: 1.3.0
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
>
> Migrating from Flink 1.1 fails for CEP, since in 1.1, the legacy 
> {{MultiplexingStreamRecordSerializer}} is used for stream  elements in the 
> serialized priority queue (via the {{PriorityQueueSerializer}}).
> In newer versions, the {{StreamElementSerializer}} is used instead. For this 
> to work, we need to implement the compatibility methods for 
> {{StreamElementSerializer}} such that it is also compatible with 
> configuration snapshots taken from the {{MultiplexingStreamRecordSerializer}}.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6844) TraversableSerializer should implement compatibility methods

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6844:
---

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/4081


> TraversableSerializer should implement compatibility methods
> 
>
> Key: FLINK-6844
> URL: https://issues.apache.org/jira/browse/FLINK-6844
> Project: Flink
>  Issue Type: Bug
>  Components: Type Serialization System
>Affects Versions: 1.3.0
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
>Priority: Blocker
>  Labels: flink-rel-1.3.1-blockers
> Fix For: 1.3.1
>
>
> The {{TraversableSerializer}} may be used as a serializer for managed state 
> and takes part in checkpointing, therefore should implement the compatibility 
> methods.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6830) Add ITTests for savepoint migration from 1.3

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6830:
---

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/4059


> Add ITTests for savepoint migration from 1.3
> 
>
> Key: FLINK-6830
> URL: https://issues.apache.org/jira/browse/FLINK-6830
> Project: Flink
>  Issue Type: Test
>  Components: State Backends, Checkpointing
>Affects Versions: 1.3.0
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
> Fix For: 1.3.1
>
>
> Already with FLINK-6763 and FLINK-6764 we'll need to change the serialization 
> formats between 1.3.0 and 1.3.x.
> We probably should add the stateful job migration ITCases for restoring from 
> Flink 1.3.x now.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink pull request #4079: [FLINK-6853] [DataStream] Let StreamRecordSerializ...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/4079


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #4081: [FLINK-6844] [scala] Implement compatibility metho...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/4081


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #4084: [FLINK-6772] [cep] Fix ordering (by timestamp) of matched...

[kl0u]

Github user kl0u commented on the issue:

https://github.com/apache/flink/pull/4084
  
Thanks @dawidwys ! I will let travis have another go on the rebased version 
and then merge.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #4072: [FLINK-6848] Update managed state docs

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/4072
  
Hi @Fokko, thanks a lot for this contribution!

The current changes look good. However, while you're on it, what do you 
think about also adding a Scala counterpart example for the other code snippets 
on the page (e.g. managed operator state, source functions)?

The changes right now are good as is, but I prefer not to merge a 
"partially documented Scala example", which may give the wrong impression that 
Scala works only for some cases. What do you think?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6848) Extend the managed state docs with a Scala example

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6848:
---

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/4072
  
Hi @Fokko, thanks a lot for this contribution!

The current changes look good. However, while you're on it, what do you 
think about also adding a Scala counterpart example for the other code snippets 
on the page (e.g. managed operator state, source functions)?

The changes right now are good as is, but I prefer not to merge a 
"partially documented Scala example", which may give the wrong impression that 
Scala works only for some cases. What do you think?


> Extend the managed state docs with a Scala example
> --
>
> Key: FLINK-6848
> URL: https://issues.apache.org/jira/browse/FLINK-6848
> Project: Flink
>  Issue Type: Bug
>Reporter: Fokko Driesprong
>
> Hi all,
> It would be nice to add a Scala example code snippet in the Managed state 
> docs. This makes it a bit easier to start using managed state in Scala. The 
> code is tested and works.
> Kind regards,
> Fokko



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6808) Stream join fails when checkpointing is enabled

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6808:
---

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/4052


> Stream join fails when checkpointing is enabled
> ---
>
> Key: FLINK-6808
> URL: https://issues.apache.org/jira/browse/FLINK-6808
> Project: Flink
>  Issue Type: Bug
>  Components: DataStream API
>Affects Versions: 1.3.0
>Reporter: Francisco Rosa
>Assignee: Tzu-Li (Gordon) Tai
>Priority: Blocker
>  Labels: flink-rel-1.3.1-blockers
> Fix For: 1.3.1
>
>
> The combination of joining streams and checkpointing fails in 1.3.0. It used 
> to work with the previous 1.2 version. Code example for failure:
> {code:title=Example|borderStyle=solid}
> public static void main(String[] args) throws Exception {
> final StreamExecutionEnvironment env = 
> StreamExecutionEnvironment.getExecutionEnvironment();
> // enable checkpoints
> env.enableCheckpointing(5000);
> // create two streams
> DataStreamSource one = env.generateSequence(0, 5000);
> DataStreamSource two = env.generateSequence(2000, 15000);
> // process both, provide a delay to make sure checkpoint will happen
> DataStream oneProcessed = one.
> map(oneValue -> {
> Thread.sleep(1000);
> return "val-" + oneValue;
> });
> DataStream twoProcessed = two.
> map(oneValue -> {
> Thread.sleep(1000);
> return "val-" + oneValue;
> });
> // join the two streams, join on string match
> DataStream joinedStreams = oneProcessed.
> join(twoProcessed).
> where(String::toString).
> equalTo(String::toString).
> window(TumblingProcessingTimeWindows.of(Time.seconds(5))).
> apply(new JoinFunction() {
> @Override
> public String join(String oneValue, String twoValue) {
> // nothing really relevant, just concatenate string
> return oneValue + "+" + twoValue;
> }
> });
> // output results
> joinedStreams.print();
> env.execute("Issue with stream join and checkpoints");
> }
> {code}
> Stack trace:
> {noformat}
> java.lang.Exception: Could not perform checkpoint 1 for operator 
> TriggerWindow(TumblingProcessingTimeWindows(5000), 
> ListStateDescriptor{serializer=org.apache.flink.api.common.typeutils.base.ListSerializer@3769cce0},
>  ProcessingTimeTrigger(), WindowedStream.apply(CoGroupedStreams.java:300)) -> 
> Sink: Unnamed (1/1).
>   at 
> org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:550)
>   at 
> org.apache.flink.streaming.runtime.io.BarrierBuffer.notifyCheckpoint(BarrierBuffer.java:378)
>   at 
> org.apache.flink.streaming.runtime.io.BarrierBuffer.processBarrier(BarrierBuffer.java:281)
>   at 
> org.apache.flink.streaming.runtime.io.BarrierBuffer.getNextNonBlocked(BarrierBuffer.java:183)
>   at 
> org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:213)
>   at 
> org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:69)
>   at 
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:262)
>   at org.apache.flink.runtime.taskmanager.Task.run(Task.java:702)
>   at java.lang.Thread.run(Thread.java:745)
> Caused by: java.lang.Exception: Could not complete snapshot 1 for operator 
> TriggerWindow(TumblingProcessingTimeWindows(5000), 
> ListStateDescriptor{serializer=org.apache.flink.api.common.typeutils.base.ListSerializer@3769cce0},
>  ProcessingTimeTrigger(), WindowedStream.apply(CoGroupedStreams.java:300)) -> 
> Sink: Unnamed (1/1).
>   at 
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:406)
>   at 
> org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.checkpointStreamOperator(StreamTask.java:1157)
>   at 
> org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.executeCheckpointing(StreamTask.java:1089)
>   at 
> org.apache.flink.streaming.runtime.tasks.StreamTask.checkpointState(StreamTask.java:653)
>   at 
> org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:589)
>   at 
> 

[jira] [Commented] (FLINK-6849) Refactor operator state backend and internal operator state hierarchy

[mingleizhang (JIRA)]

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

mingleizhang commented on FLINK-6849:
-

I think we can do this refactor by the below three steps.

1. Refactor {{OperatorStateBackend}} to having a more functionality, add the 
function of create and cache already registered state, such as add 
{{getOrCreateOperatorState}} and {{getPartitionedState}} methods to 
{{OperatorStateBackend}} interface.
2. Add abstract class implements for {{OperatorStateBackend}}. For example, we 
can have a class called {{AbstractOperatorStateBackend}}.
3. Remove the concrete cache operation for {{DefaultOperatorStateBackend}} and 
then implement that in {{AbstractOperatorStateBackend}}.

[~tzulitai], What do you think ? It would be helpful if you can take a look at 
those steps.

> Refactor operator state backend and internal operator state hierarchy
> -
>
> Key: FLINK-6849
> URL: https://issues.apache.org/jira/browse/FLINK-6849
> Project: Flink
>  Issue Type: Improvement
>  Components: State Backends, Checkpointing
>Reporter: Tzu-Li (Gordon) Tai
>
> Currently, compared to the keyed state backends, the operator state backends, 
> as well as operator state interfaces, lacks proper hierarchy.
> One issue with this lack of hierarchy is that the general concerns of 
> implementing state registration is different between the keyed and operator 
> backends (aside from what is naturally different, such as namespace and key 
> which is not relevant for the operator backend). For example, in the keyed 
> backend hierarchy, {{AbstractKeyedStateBackend}} has caches that shortcuts 
> re-accessing already registered state. This behaviour is missing in the 
> operator backend hierarchy, and for example needs to be explicitly handled by 
> the concrete {{DefaultOperatorStateBackend}} subclass implementation.
> As of now, the need of a proper hierarchy also on the operator backend side 
> might not be that prominent, but will mostly likely become more prominent  as 
> we wish to introduce more state structures for operator state (e.g. a 
> {{MapState}} for operator state has already been discussed a few times 
> already) as well as more options besides memory-backed operator state.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Comment Edited] (FLINK-6849) Refactor operator state backend and internal operator state hierarchy

[mingleizhang (JIRA)]

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

mingleizhang edited comment on FLINK-6849 at 6/7/17 7:03 AM:
-

I think we can do this refactor by the below three steps.

1. Refactor {{OperatorStateBackend}} to have a more functionality, add the 
function of create and cache already registered state, such as add 
{{getOrCreateOperatorState}} and {{getPartitionedState}} methods to 
{{OperatorStateBackend}} interface.
2. Add abstract class implements for {{OperatorStateBackend}}. For example, we 
can have a class called {{AbstractOperatorStateBackend}}.
3. Remove the concrete cache operation for {{DefaultOperatorStateBackend}} and 
then implement that in {{AbstractOperatorStateBackend}}.

[~tzulitai], What do you think ? It would be helpful if you can take a look at 
those steps.


was (Author: mingleizhang):
I think that we can do this refactor by the below three steps.

1. Refactor {{OperatorStateBackend}} to have a more functionality, add the 
function of create and cache already registered state, such as add 
{{getOrCreateOperatorState}} and {{getPartitionedState}} methods to 
{{OperatorStateBackend}} interface.
2. Add abstract class implements for {{OperatorStateBackend}}. For example, we 
can have a class called {{AbstractOperatorStateBackend}}.
3. Remove the concrete cache operation for {{DefaultOperatorStateBackend}} and 
then implement that in {{AbstractOperatorStateBackend}}.

[~tzulitai], What do you think ? It would be helpful if you can take a look at 
those steps.

> Refactor operator state backend and internal operator state hierarchy
> -
>
> Key: FLINK-6849
> URL: https://issues.apache.org/jira/browse/FLINK-6849
> Project: Flink
>  Issue Type: Improvement
>  Components: State Backends, Checkpointing
>Reporter: Tzu-Li (Gordon) Tai
>
> Currently, compared to the keyed state backends, the operator state backends, 
> as well as operator state interfaces, lacks proper hierarchy.
> One issue with this lack of hierarchy is that the general concerns of 
> implementing state registration is different between the keyed and operator 
> backends (aside from what is naturally different, such as namespace and key 
> which is not relevant for the operator backend). For example, in the keyed 
> backend hierarchy, {{AbstractKeyedStateBackend}} has caches that shortcuts 
> re-accessing already registered state. This behaviour is missing in the 
> operator backend hierarchy, and for example needs to be explicitly handled by 
> the concrete {{DefaultOperatorStateBackend}} subclass implementation.
> As of now, the need of a proper hierarchy also on the operator backend side 
> might not be that prominent, but will mostly likely become more prominent  as 
> we wish to introduce more state structures for operator state (e.g. a 
> {{MapState}} for operator state has already been discussed a few times 
> already) as well as more options besides memory-backed operator state.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Comment Edited] (FLINK-6849) Refactor operator state backend and internal operator state hierarchy

[mingleizhang (JIRA)]

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

mingleizhang edited comment on FLINK-6849 at 6/7/17 6:41 AM:
-

I think we can do this refactor by the below three steps.

1. Refactor {{OperatorStateBackend}} to have a more functionality, add the 
function of create and cache already registered state, such as add 
{{getOrCreateOperatorState}} and {{getPartitionedState}} methods to 
{{OperatorStateBackend}} interface.
2. Add abstract class implements for {{OperatorStateBackend}}. For example, we 
can have a class called {{AbstractOperatorStateBackend}}.
3. Remove the concrete cache operation for {{DefaultOperatorStateBackend}} and 
then implement that in {{AbstractOperatorStateBackend}}.

[~tzulitai], What do you think ? It would be helpful if you can take a look at 
those steps.


was (Author: mingleizhang):
I think we can do this refactor by the below three steps.

1. Refactor {{OperatorStateBackend}} to having a more functionality, add the 
function of create and cache already registered state, such as add 
{{getOrCreateOperatorState}} and {{getPartitionedState}} methods to 
{{OperatorStateBackend}} interface.
2. Add abstract class implements for {{OperatorStateBackend}}. For example, we 
can have a class called {{AbstractOperatorStateBackend}}.
3. Remove the concrete cache operation for {{DefaultOperatorStateBackend}} and 
then implement that in {{AbstractOperatorStateBackend}}.

[~tzulitai], What do you think ? It would be helpful if you can take a look at 
those steps.

> Refactor operator state backend and internal operator state hierarchy
> -
>
> Key: FLINK-6849
> URL: https://issues.apache.org/jira/browse/FLINK-6849
> Project: Flink
>  Issue Type: Improvement
>  Components: State Backends, Checkpointing
>Reporter: Tzu-Li (Gordon) Tai
>
> Currently, compared to the keyed state backends, the operator state backends, 
> as well as operator state interfaces, lacks proper hierarchy.
> One issue with this lack of hierarchy is that the general concerns of 
> implementing state registration is different between the keyed and operator 
> backends (aside from what is naturally different, such as namespace and key 
> which is not relevant for the operator backend). For example, in the keyed 
> backend hierarchy, {{AbstractKeyedStateBackend}} has caches that shortcuts 
> re-accessing already registered state. This behaviour is missing in the 
> operator backend hierarchy, and for example needs to be explicitly handled by 
> the concrete {{DefaultOperatorStateBackend}} subclass implementation.
> As of now, the need of a proper hierarchy also on the operator backend side 
> might not be that prominent, but will mostly likely become more prominent  as 
> we wish to introduce more state structures for operator state (e.g. a 
> {{MapState}} for operator state has already been discussed a few times 
> already) as well as more options besides memory-backed operator state.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink issue #3998: [FLINK-6661][web] Merge "Subtasks" and "SubtasksByTaskMan...

[zentol]

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/3998
  
we got enough space, so why not be explicit about it :)

Will change it while merging.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Comment Edited] (FLINK-6849) Refactor operator state backend and internal operator state hierarchy

[mingleizhang (JIRA)]

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

mingleizhang edited comment on FLINK-6849 at 6/7/17 6:59 AM:
-

I think that we can do this refactor by the below three steps.

1. Refactor {{OperatorStateBackend}} to have a more functionality, add the 
function of create and cache already registered state, such as add 
{{getOrCreateOperatorState}} and {{getPartitionedState}} methods to 
{{OperatorStateBackend}} interface.
2. Add abstract class implements for {{OperatorStateBackend}}. For example, we 
can have a class called {{AbstractOperatorStateBackend}}.
3. Remove the concrete cache operation for {{DefaultOperatorStateBackend}} and 
then implement that in {{AbstractOperatorStateBackend}}.

[~tzulitai], What do you think ? It would be helpful if you can take a look at 
those steps.


was (Author: mingleizhang):
I think we can do this refactor by the below three steps.

1. Refactor {{OperatorStateBackend}} to have a more functionality, add the 
function of create and cache already registered state, such as add 
{{getOrCreateOperatorState}} and {{getPartitionedState}} methods to 
{{OperatorStateBackend}} interface.
2. Add abstract class implements for {{OperatorStateBackend}}. For example, we 
can have a class called {{AbstractOperatorStateBackend}}.
3. Remove the concrete cache operation for {{DefaultOperatorStateBackend}} and 
then implement that in {{AbstractOperatorStateBackend}}.

[~tzulitai], What do you think ? It would be helpful if you can take a look at 
those steps.

> Refactor operator state backend and internal operator state hierarchy
> -
>
> Key: FLINK-6849
> URL: https://issues.apache.org/jira/browse/FLINK-6849
> Project: Flink
>  Issue Type: Improvement
>  Components: State Backends, Checkpointing
>Reporter: Tzu-Li (Gordon) Tai
>
> Currently, compared to the keyed state backends, the operator state backends, 
> as well as operator state interfaces, lacks proper hierarchy.
> One issue with this lack of hierarchy is that the general concerns of 
> implementing state registration is different between the keyed and operator 
> backends (aside from what is naturally different, such as namespace and key 
> which is not relevant for the operator backend). For example, in the keyed 
> backend hierarchy, {{AbstractKeyedStateBackend}} has caches that shortcuts 
> re-accessing already registered state. This behaviour is missing in the 
> operator backend hierarchy, and for example needs to be explicitly handled by 
> the concrete {{DefaultOperatorStateBackend}} subclass implementation.
> As of now, the need of a proper hierarchy also on the operator backend side 
> might not be that prominent, but will mostly likely become more prominent  as 
> we wish to introduce more state structures for operator state (e.g. a 
> {{MapState}} for operator state has already been discussed a few times 
> already) as well as more options besides memory-backed operator state.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6661) Merge "Subtasks" and "TaskManagers" view

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6661:
---

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/3998
  
we got enough space, so why not be explicit about it :)

Will change it while merging.


> Merge "Subtasks" and "TaskManagers" view
> 
>
> Key: FLINK-6661
> URL: https://issues.apache.org/jira/browse/FLINK-6661
> Project: Flink
>  Issue Type: Sub-task
>  Components: Webfrontend
>Affects Versions: 1.4.0
>Reporter: Chesnay Schepler
>Assignee: Chesnay Schepler
>
> The {{Subtasks}} and {{TaskManagers/ Subtasks by TaskManager}} view are very 
> similar in what they do, so much in fact that they are identical if the 
> taskmanagers have a single task slot.
> I propose to merge them, and add a checkbox to aggregate the subtasks by 
> taskmanagers.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Resolved] (FLINK-6844) TraversableSerializer should implement compatibility methods

[Tzu-Li (Gordon) Tai (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6844?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Tzu-Li (Gordon) Tai resolved FLINK-6844.

   Resolution: Fixed
Fix Version/s: 1.4.0

Fixed for 1.3 via e1e207c898ed436df656d01364cf0e5fa818b730.
Fixed for master via c11d5ed5388a5a30ca4ea0c5ac68e22e5989cb54.

> TraversableSerializer should implement compatibility methods
> 
>
> Key: FLINK-6844
> URL: https://issues.apache.org/jira/browse/FLINK-6844
> Project: Flink
>  Issue Type: Bug
>  Components: Type Serialization System
>Affects Versions: 1.3.0
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
>Priority: Blocker
>  Labels: flink-rel-1.3.1-blockers
> Fix For: 1.3.1, 1.4.0
>
>
> The {{TraversableSerializer}} may be used as a serializer for managed state 
> and takes part in checkpointing, therefore should implement the compatibility 
> methods.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Closed] (FLINK-6844) TraversableSerializer should implement compatibility methods

[Tzu-Li (Gordon) Tai (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6844?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Tzu-Li (Gordon) Tai closed FLINK-6844.
--

> TraversableSerializer should implement compatibility methods
> 
>
> Key: FLINK-6844
> URL: https://issues.apache.org/jira/browse/FLINK-6844
> Project: Flink
>  Issue Type: Bug
>  Components: Type Serialization System
>Affects Versions: 1.3.0
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
>Priority: Blocker
>  Labels: flink-rel-1.3.1-blockers
> Fix For: 1.3.1, 1.4.0
>
>
> The {{TraversableSerializer}} may be used as a serializer for managed state 
> and takes part in checkpointing, therefore should implement the compatibility 
> methods.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink issue #4086: [FLINK-6865] Update checkstyle documentation

[zentol]

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4086
  
hmmi just retried it and it didn't work this time. Let me investigate a 
bit.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6865) Expand checkstyle docs to include import in intellij

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6865:
---

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4086
  
hmmi just retried it and it didn't work this time. Let me investigate a 
bit.


> Expand checkstyle docs to include import in intellij
> 
>
> Key: FLINK-6865
> URL: https://issues.apache.org/jira/browse/FLINK-6865
> Project: Flink
>  Issue Type: Improvement
>  Components: Checkstyle
>Affects Versions: 1.4.0
>Reporter: Chesnay Schepler
>Assignee: Chesnay Schepler
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Closed] (FLINK-6853) Migrating from Flink 1.1 fails for FlinkCEP

[Tzu-Li (Gordon) Tai (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6853?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Tzu-Li (Gordon) Tai closed FLINK-6853.
--

> Migrating from Flink 1.1 fails for FlinkCEP
> ---
>
> Key: FLINK-6853
> URL: https://issues.apache.org/jira/browse/FLINK-6853
> Project: Flink
>  Issue Type: Bug
>  Components: CEP
>Affects Versions: 1.3.0
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
> Fix For: 1.3.1, 1.4.0
>
>
> Migrating from Flink 1.1 fails for CEP, since in 1.1, the legacy 
> {{MultiplexingStreamRecordSerializer}} is used for stream  elements in the 
> serialized priority queue (via the {{PriorityQueueSerializer}}).
> In newer versions, the {{StreamElementSerializer}} is used instead. For this 
> to work, we need to implement the compatibility methods for 
> {{StreamElementSerializer}} such that it is also compatible with 
> configuration snapshots taken from the {{MultiplexingStreamRecordSerializer}}.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink issue #4086: [FLINK-6865] Update checkstyle documentation

[zentol]

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4086
  
Something the import did in fact set was spaces around operators. It may be 
that not all checkstyle rules are properly imported.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6865) Expand checkstyle docs to include import in intellij

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6865:
---

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4086
  
Something the import did in fact set was spaces around operators. It may be 
that not all checkstyle rules are properly imported.


> Expand checkstyle docs to include import in intellij
> 
>
> Key: FLINK-6865
> URL: https://issues.apache.org/jira/browse/FLINK-6865
> Project: Flink
>  Issue Type: Improvement
>  Components: Checkstyle
>Affects Versions: 1.4.0
>Reporter: Chesnay Schepler
>Assignee: Chesnay Schepler
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Resolved] (FLINK-6830) Add ITTests for savepoint migration from 1.3

[Tzu-Li (Gordon) Tai (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6830?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Tzu-Li (Gordon) Tai resolved FLINK-6830.

   Resolution: Fixed
Fix Version/s: 1.4.0

Fixed for master via 3792be4b5f80826d5dbf51c0517e8c00847472ca.
Fixed for 1.3.1 via d4a646a035366918a100f64428c471464870b8d0.

> Add ITTests for savepoint migration from 1.3
> 
>
> Key: FLINK-6830
> URL: https://issues.apache.org/jira/browse/FLINK-6830
> Project: Flink
>  Issue Type: Test
>  Components: State Backends, Checkpointing
>Affects Versions: 1.3.0
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
> Fix For: 1.3.1, 1.4.0
>
>
> Already with FLINK-6763 and FLINK-6764 we'll need to change the serialization 
> formats between 1.3.0 and 1.3.x.
> We probably should add the stateful job migration ITCases for restoring from 
> Flink 1.3.x now.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink issue #4086: [FLINK-6865] Update checkstyle documentation

[zentol]

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4086
  
yes, it's a good idea to "flip" the checkstyle names.

I only checked a few things, but importing the checkstyle configuration, 
with the Checkstyle-IDEA plugin enabled, at the very least correctly configured 
the import order. 


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6865) Expand checkstyle docs to include import in intellij

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6865:
---

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4086
  
yes, it's a good idea to "flip" the checkstyle names.

I only checked a few things, but importing the checkstyle configuration, 
with the Checkstyle-IDEA plugin enabled, at the very least correctly configured 
the import order. 


> Expand checkstyle docs to include import in intellij
> 
>
> Key: FLINK-6865
> URL: https://issues.apache.org/jira/browse/FLINK-6865
> Project: Flink
>  Issue Type: Improvement
>  Components: Checkstyle
>Affects Versions: 1.4.0
>Reporter: Chesnay Schepler
>Assignee: Chesnay Schepler
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Resolved] (FLINK-6815) Javadocs don't work anymore in Flink 1.4-SNAPSHOT

[Robert Metzger (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6815?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Robert Metzger resolved FLINK-6815.
---
   Resolution: Fixed
Fix Version/s: 1.4.0
   1.3.1

1.4: http://git-wip-us.apache.org/repos/asf/flink/commit/e13a7f80
1.3: http://git-wip-us.apache.org/repos/asf/flink/commit/f72eff7f

> Javadocs don't work anymore in Flink 1.4-SNAPSHOT
> -
>
> Key: FLINK-6815
> URL: https://issues.apache.org/jira/browse/FLINK-6815
> Project: Flink
>  Issue Type: Bug
>  Components: Build System
>Affects Versions: 1.4.0
>Reporter: Robert Metzger
>Assignee: Robert Metzger
> Fix For: 1.3.1, 1.4.0
>
>
> https://ci.apache.org/projects/flink/flink-docs-master/api/java/org/apache/flink/streaming/api/scala/KeyedStream.html
>  results in a 404 error.
> The problem 
> (https://ci.apache.org/builders/flink-docs-master/builds/731/steps/Java%20&%20Scala%20docs/logs/stdio)
>  is the following:
> {code}
> [ERROR] Failed to execute goal 
> net.alchim31.maven:scala-maven-plugin:3.2.2:compile (doc) on project 
> flink-annotations: wrap: 
> org.apache.maven.artifact.resolver.ArtifactNotFoundException: Could not find 
> artifact com.typesafe.genjavadoc:genjavadoc-plugin_2.10.6:jar:0.8 in central 
> (https://repo.maven.apache.org/maven2)
> {code}
> I think the problem is that we upgraded the scala version to 2.10.6, but the 
> plugin doesn't have version 0.8 for that scala version.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Resolved] (FLINK-6853) Migrating from Flink 1.1 fails for FlinkCEP

[Tzu-Li (Gordon) Tai (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6853?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Tzu-Li (Gordon) Tai resolved FLINK-6853.

   Resolution: Fixed
Fix Version/s: 1.4.0
   1.3.1

Fixed for master via 4895472ba2279d2982a45279d7be76bf3dfd8768.
Fixed for 1.3.1 via 1d89dd06c1f9b09420ad3ff095d0842b4a951938.

> Migrating from Flink 1.1 fails for FlinkCEP
> ---
>
> Key: FLINK-6853
> URL: https://issues.apache.org/jira/browse/FLINK-6853
> Project: Flink
>  Issue Type: Bug
>  Components: CEP
>Affects Versions: 1.3.0
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
> Fix For: 1.3.1, 1.4.0
>
>
> Migrating from Flink 1.1 fails for CEP, since in 1.1, the legacy 
> {{MultiplexingStreamRecordSerializer}} is used for stream  elements in the 
> serialized priority queue (via the {{PriorityQueueSerializer}}).
> In newer versions, the {{StreamElementSerializer}} is used instead. For this 
> to work, we need to implement the compatibility methods for 
> {{StreamElementSerializer}} such that it is also compatible with 
> configuration snapshots taken from the {{MultiplexingStreamRecordSerializer}}.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Resolved] (FLINK-6808) Stream join fails when checkpointing is enabled

[Tzu-Li (Gordon) Tai (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6808?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Tzu-Li (Gordon) Tai resolved FLINK-6808.

Resolution: Fixed

Fixed for 1.3.1 via f74caf7062b1cc23a704f8f8b8171be430b60807.
Fixed for master via 539787b21822eb839d0408a989cd541450bd08d2.

> Stream join fails when checkpointing is enabled
> ---
>
> Key: FLINK-6808
> URL: https://issues.apache.org/jira/browse/FLINK-6808
> Project: Flink
>  Issue Type: Bug
>  Components: DataStream API
>Affects Versions: 1.3.0
>Reporter: Francisco Rosa
>Assignee: Tzu-Li (Gordon) Tai
>Priority: Blocker
>  Labels: flink-rel-1.3.1-blockers
> Fix For: 1.3.1
>
>
> The combination of joining streams and checkpointing fails in 1.3.0. It used 
> to work with the previous 1.2 version. Code example for failure:
> {code:title=Example|borderStyle=solid}
> public static void main(String[] args) throws Exception {
> final StreamExecutionEnvironment env = 
> StreamExecutionEnvironment.getExecutionEnvironment();
> // enable checkpoints
> env.enableCheckpointing(5000);
> // create two streams
> DataStreamSource one = env.generateSequence(0, 5000);
> DataStreamSource two = env.generateSequence(2000, 15000);
> // process both, provide a delay to make sure checkpoint will happen
> DataStream oneProcessed = one.
> map(oneValue -> {
> Thread.sleep(1000);
> return "val-" + oneValue;
> });
> DataStream twoProcessed = two.
> map(oneValue -> {
> Thread.sleep(1000);
> return "val-" + oneValue;
> });
> // join the two streams, join on string match
> DataStream joinedStreams = oneProcessed.
> join(twoProcessed).
> where(String::toString).
> equalTo(String::toString).
> window(TumblingProcessingTimeWindows.of(Time.seconds(5))).
> apply(new JoinFunction() {
> @Override
> public String join(String oneValue, String twoValue) {
> // nothing really relevant, just concatenate string
> return oneValue + "+" + twoValue;
> }
> });
> // output results
> joinedStreams.print();
> env.execute("Issue with stream join and checkpoints");
> }
> {code}
> Stack trace:
> {noformat}
> java.lang.Exception: Could not perform checkpoint 1 for operator 
> TriggerWindow(TumblingProcessingTimeWindows(5000), 
> ListStateDescriptor{serializer=org.apache.flink.api.common.typeutils.base.ListSerializer@3769cce0},
>  ProcessingTimeTrigger(), WindowedStream.apply(CoGroupedStreams.java:300)) -> 
> Sink: Unnamed (1/1).
>   at 
> org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:550)
>   at 
> org.apache.flink.streaming.runtime.io.BarrierBuffer.notifyCheckpoint(BarrierBuffer.java:378)
>   at 
> org.apache.flink.streaming.runtime.io.BarrierBuffer.processBarrier(BarrierBuffer.java:281)
>   at 
> org.apache.flink.streaming.runtime.io.BarrierBuffer.getNextNonBlocked(BarrierBuffer.java:183)
>   at 
> org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:213)
>   at 
> org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:69)
>   at 
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:262)
>   at org.apache.flink.runtime.taskmanager.Task.run(Task.java:702)
>   at java.lang.Thread.run(Thread.java:745)
> Caused by: java.lang.Exception: Could not complete snapshot 1 for operator 
> TriggerWindow(TumblingProcessingTimeWindows(5000), 
> ListStateDescriptor{serializer=org.apache.flink.api.common.typeutils.base.ListSerializer@3769cce0},
>  ProcessingTimeTrigger(), WindowedStream.apply(CoGroupedStreams.java:300)) -> 
> Sink: Unnamed (1/1).
>   at 
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:406)
>   at 
> org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.checkpointStreamOperator(StreamTask.java:1157)
>   at 
> org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.executeCheckpointing(StreamTask.java:1089)
>   at 
> org.apache.flink.streaming.runtime.tasks.StreamTask.checkpointState(StreamTask.java:653)
>   at 
> org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:589)
>   at 
> 

[jira] [Closed] (FLINK-6830) Add ITTests for savepoint migration from 1.3

[Tzu-Li (Gordon) Tai (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6830?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Tzu-Li (Gordon) Tai closed FLINK-6830.
--

> Add ITTests for savepoint migration from 1.3
> 
>
> Key: FLINK-6830
> URL: https://issues.apache.org/jira/browse/FLINK-6830
> Project: Flink
>  Issue Type: Test
>  Components: State Backends, Checkpointing
>Affects Versions: 1.3.0
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
> Fix For: 1.3.1, 1.4.0
>
>
> Already with FLINK-6763 and FLINK-6764 we'll need to change the serialization 
> formats between 1.3.0 and 1.3.x.
> We probably should add the stateful job migration ITCases for restoring from 
> Flink 1.3.x now.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6488) Remove 'start-local.sh' script

[Chesnay Schepler (JIRA)]

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

Chesnay Schepler commented on FLINK-6488:
-

Of course it is redundant, that's the whole point.

If we remove the script we're gonna break someone's workflow all of a sudden; i 
know that I would have to modify some of my own scripts.

There's no harm in keeping it. We can also change it now to call 
{{start-cluster.sh}} and print a warning that it will be removed in 1.5, so 
that we at least give people some time to adjust.

> Remove 'start-local.sh' script
> --
>
> Key: FLINK-6488
> URL: https://issues.apache.org/jira/browse/FLINK-6488
> Project: Flink
>  Issue Type: Sub-task
>  Components: Startup Shell Scripts
>Reporter: Stephan Ewen
>Assignee: mingleizhang
>
> The {{start-cluster.sh}} scripts work locally now, without needing SSH setup.
> We can remove {{start-local.sh}} without any loss of functionality.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6488) Remove 'start-local.sh' script

[mingleizhang (JIRA)]

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

mingleizhang commented on FLINK-6488:
-

[~Zentol] So, We will still keep {{start-local.sh}} and do not remove it until 
1.5 version of Flink. At this moment for this issue, we could just refine 
{{start-cluster.sh}} with print a warning message to people. What do you think ?

> Remove 'start-local.sh' script
> --
>
> Key: FLINK-6488
> URL: https://issues.apache.org/jira/browse/FLINK-6488
> Project: Flink
>  Issue Type: Sub-task
>  Components: Startup Shell Scripts
>Reporter: Stephan Ewen
>Assignee: mingleizhang
>
> The {{start-cluster.sh}} scripts work locally now, without needing SSH setup.
> We can remove {{start-local.sh}} without any loss of functionality.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6488) Remove 'start-local.sh' script

[mingleizhang (JIRA)]

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

mingleizhang commented on FLINK-6488:
-

That makes sense better. Patch will available soon.

> Remove 'start-local.sh' script
> --
>
> Key: FLINK-6488
> URL: https://issues.apache.org/jira/browse/FLINK-6488
> Project: Flink
>  Issue Type: Sub-task
>  Components: Startup Shell Scripts
>Reporter: Stephan Ewen
>Assignee: mingleizhang
>
> The {{start-cluster.sh}} scripts work locally now, without needing SSH setup.
> We can remove {{start-local.sh}} without any loss of functionality.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6281) Create TableSink for JDBC

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6281:
---

Github user zentol commented on a diff in the pull request:

https://github.com/apache/flink/pull/3712#discussion_r120576118
  
--- Diff: 
flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCOutputFormat.java
 ---
@@ -202,14 +202,20 @@ public void writeRecord(Row row) throws IOException {
upload.addBatch();
batchCount++;
if (batchCount >= batchInterval) {
-   upload.executeBatch();
-   batchCount = 0;
+   flush();
}
} catch (SQLException | IllegalArgumentException e) {
throw new IllegalArgumentException("writeRecord() 
failed", e);
}
}
 
+   void flush() throws SQLException {
+   if (upload != null) {
+   upload.executeBatch();
--- End diff --

It's been a while since i worked with JDBC, I take it this is a synchronous 
call? What happens if this call fails?


> Create TableSink for JDBC
> -
>
> Key: FLINK-6281
> URL: https://issues.apache.org/jira/browse/FLINK-6281
> Project: Flink
>  Issue Type: Improvement
>  Components: Table API & SQL
>Reporter: Haohui Mai
>Assignee: Haohui Mai
>
> It would be nice to integrate the table APIs with the JDBC connectors so that 
> the rows in the tables can be directly pushed into JDBC.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink pull request #3712: [FLINK-6281] Create TableSink for JDBC.

[zentol]

Github user zentol commented on a diff in the pull request:

https://github.com/apache/flink/pull/3712#discussion_r120575757
  
--- Diff: 
flink-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JDBCTableSinkTest.java
 ---
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.java.io.jdbc;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotSame;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Test for JDBCTableSink.
+ */
+public class JDBCTableSinkTest {
+   private static final String[] FIELD_NAMES = new String[]{"foo"};
+   private static final TypeInformation[] FIELD_TYPES = new 
TypeInformation[]{
+   BasicTypeInfo.STRING_TYPE_INFO
+   };
+
+   @Test
+   public void testOutputSink() throws Exception {
+   JDBCOutputFormat outputFormat = mock(JDBCOutputFormat.class);
+   JDBCTableSink sink = new JDBCTableSink(outputFormat);
+   @SuppressWarnings("unchecked")
+   DataStream dataStream = (DataStream) 
mock(DataStream.class);
+   sink.emitDataStream(dataStream);
+   verify(dataStream).addSink(sink);
+   }
+
+   @Test
+   public void testFlush() throws Exception {
+   JDBCOutputFormat outputFormat = mock(JDBCOutputFormat.class);
+   JDBCTableSink sink = new JDBCTableSink(outputFormat);
+   @SuppressWarnings("unchecked")
+   DataStream dataStream = (DataStream) 
mock(DataStream.class);
+   sink.emitDataStream(dataStream);
+   sink.snapshotState(mock(FunctionSnapshotContext.class));
+   verify(dataStream).addSink(sink);
+   verify(outputFormat).flush();
--- End diff --

let's not use mocking for this test. Just create an actual format/sink, 
give N values to the sink where N < batchSize, verify they haven't been written 
yet, call flush, verify they were written.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6281) Create TableSink for JDBC

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6281:
---

Github user zentol commented on a diff in the pull request:

https://github.com/apache/flink/pull/3712#discussion_r120575757
  
--- Diff: 
flink-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JDBCTableSinkTest.java
 ---
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.java.io.jdbc;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotSame;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Test for JDBCTableSink.
+ */
+public class JDBCTableSinkTest {
+   private static final String[] FIELD_NAMES = new String[]{"foo"};
+   private static final TypeInformation[] FIELD_TYPES = new 
TypeInformation[]{
+   BasicTypeInfo.STRING_TYPE_INFO
+   };
+
+   @Test
+   public void testOutputSink() throws Exception {
+   JDBCOutputFormat outputFormat = mock(JDBCOutputFormat.class);
+   JDBCTableSink sink = new JDBCTableSink(outputFormat);
+   @SuppressWarnings("unchecked")
+   DataStream dataStream = (DataStream) 
mock(DataStream.class);
+   sink.emitDataStream(dataStream);
+   verify(dataStream).addSink(sink);
+   }
+
+   @Test
+   public void testFlush() throws Exception {
+   JDBCOutputFormat outputFormat = mock(JDBCOutputFormat.class);
+   JDBCTableSink sink = new JDBCTableSink(outputFormat);
+   @SuppressWarnings("unchecked")
+   DataStream dataStream = (DataStream) 
mock(DataStream.class);
+   sink.emitDataStream(dataStream);
+   sink.snapshotState(mock(FunctionSnapshotContext.class));
+   verify(dataStream).addSink(sink);
+   verify(outputFormat).flush();
--- End diff --

let's not use mocking for this test. Just create an actual format/sink, 
give N values to the sink where N < batchSize, verify they haven't been written 
yet, call flush, verify they were written.


> Create TableSink for JDBC
> -
>
> Key: FLINK-6281
> URL: https://issues.apache.org/jira/browse/FLINK-6281
> Project: Flink
>  Issue Type: Improvement
>  Components: Table API & SQL
>Reporter: Haohui Mai
>Assignee: Haohui Mai
>
> It would be nice to integrate the table APIs with the JDBC connectors so that 
> the rows in the tables can be directly pushed into JDBC.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink pull request #3712: [FLINK-6281] Create TableSink for JDBC.

[zentol]

Github user zentol commented on a diff in the pull request:

https://github.com/apache/flink/pull/3712#discussion_r120576032
  
--- Diff: 
flink-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JDBCTableSinkTest.java
 ---
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.java.io.jdbc;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotSame;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Test for JDBCTableSink.
+ */
+public class JDBCTableSinkTest {
+   private static final String[] FIELD_NAMES = new String[]{"foo"};
+   private static final TypeInformation[] FIELD_TYPES = new 
TypeInformation[]{
+   BasicTypeInfo.STRING_TYPE_INFO
+   };
+
+   @Test
+   public void testOutputSink() throws Exception {
+   JDBCOutputFormat outputFormat = mock(JDBCOutputFormat.class);
+   JDBCTableSink sink = new JDBCTableSink(outputFormat);
+   @SuppressWarnings("unchecked")
+   DataStream dataStream = (DataStream) 
mock(DataStream.class);
+   sink.emitDataStream(dataStream);
+   verify(dataStream).addSink(sink);
--- End diff --

you don't have to test this, as it is not a detail of the JDBCTableSink but 
the table API.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6488) Remove 'start-local.sh' script

[mingleizhang (JIRA)]

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

mingleizhang commented on FLINK-6488:
-

It would seems redundant if we keep {{start-local.sh}}. But If we use 
{{start-cluster.sh}} for local development and testing looks wierd too. So, I 
guess that we just rename the {{start-cluster.sh}} to 
{{start-local-cluster.sh}} probably be nice. What do you think ? :P

> Remove 'start-local.sh' script
> --
>
> Key: FLINK-6488
> URL: https://issues.apache.org/jira/browse/FLINK-6488
> Project: Flink
>  Issue Type: Sub-task
>  Components: Startup Shell Scripts
>Reporter: Stephan Ewen
>Assignee: mingleizhang
>
> The {{start-cluster.sh}} scripts work locally now, without needing SSH setup.
> We can remove {{start-local.sh}} without any loss of functionality.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Closed] (FLINK-6858) Unbounded event time Over Window emits incorrect timestamps

[Fabian Hueske (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6858?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Fabian Hueske closed FLINK-6858.

Resolution: Invalid

Hi [~Yuhong_kyo], you are of course right. The operators work exactly as they 
should and they are actually also covered by test cases 
(OverWindowHarnessTest). 
I did not look close enough when I created the issue. My apologies for that.

Closing as invalid.

> Unbounded event time Over Window emits incorrect timestamps
> ---
>
> Key: FLINK-6858
> URL: https://issues.apache.org/jira/browse/FLINK-6858
> Project: Flink
>  Issue Type: Bug
>  Components: Table API & SQL
>Affects Versions: 1.3.0
>Reporter: Fabian Hueske
>Priority: Critical
>
> The unbounded event time OVER windows emit records with incorrect timestamps.
> OVER aggregates "enrich" each input row with aggregates computed over 
> neighboring rows, i.e., they produce one output row for each input row. The 
> (event-time) timestamp of each input row should be preserved and not modified.
> All OVER window aggregates are implemented using the {{ProcessFunction}} 
> interface. The interface has two methods {{processElement()}} and 
> {{onTimer()}} that can produce output records. Records emitted by 
> {{processElement()}} are emitted with the timestamp of the record that was 
> given as an argument to the method. Records emitted by {{onTimer()}} are 
> emitted with the timestamp of the timer that triggered the call of the method.
> The implementation of the unbounded event-time OVER window registers a new 
> new timer when {{processElement()}} is called for {{currentWatermark + 1}}. 
> When the timer triggers, the {{onTimer()}} processes all rows that where 
> received between this and the last {{onTimer()}} call with timestamps smaller 
> than the current watermark. However, this means that all emitted rows have a 
> timestamp of {{currentWatermark + 1}} which is not what we want.
> The bounded event-time OVER window operators follow a different strategy and 
> register a timer for the timestamp of each row that was processed by 
> {{processElement()}} and emit the corresponding rows when {{onTimer()}} is 
> called. Hence, they emit the rows with correct timestamps.
> I think we should change the implementation of the unbounded event-time OVER 
> aggregates to a similar strategy as the bounded event-time OVER aggregates.
> What do you think [~Yuhong_kyo] [~sunjincheng121]?



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6488) Remove 'start-local.sh' script

[Chesnay Schepler (JIRA)]

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

Chesnay Schepler commented on FLINK-6488:
-

We will change {{start-local.sh}} to a) print a deprecation warning and b) call 
{{start-cluster.sh}}.

{{start-cluster.sh}} will remain unchanged.

> Remove 'start-local.sh' script
> --
>
> Key: FLINK-6488
> URL: https://issues.apache.org/jira/browse/FLINK-6488
> Project: Flink
>  Issue Type: Sub-task
>  Components: Startup Shell Scripts
>Reporter: Stephan Ewen
>Assignee: mingleizhang
>
> The {{start-cluster.sh}} scripts work locally now, without needing SSH setup.
> We can remove {{start-local.sh}} without any loss of functionality.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6849) Refactor operator state backend and internal operator state hierarchy

[Stefan Richter (JIRA)]

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

Stefan Richter commented on FLINK-6849:
---

I agree in general, because this was clear from the beginning that we need 
might have other implementations of the `OperatorStateBackend` interface. 
However, I think we should only start doing this change when we already have a 
concrete plan to introduce more implementations, e.g. based on RocksDB. I would 
also suggest that before starting this, we also spend some thoughts on the 
relationship of operator state and keyed state in general and if they could not 
be based on one common backend abstraction. For now, is there already any 
concrete need to do changes (other than cosmetical)? 

> Refactor operator state backend and internal operator state hierarchy
> -
>
> Key: FLINK-6849
> URL: https://issues.apache.org/jira/browse/FLINK-6849
> Project: Flink
>  Issue Type: Improvement
>  Components: State Backends, Checkpointing
>Reporter: Tzu-Li (Gordon) Tai
>
> Currently, compared to the keyed state backends, the operator state backends, 
> as well as operator state interfaces, lacks proper hierarchy.
> One issue with this lack of hierarchy is that the general concerns of 
> implementing state registration is different between the keyed and operator 
> backends (aside from what is naturally different, such as namespace and key 
> which is not relevant for the operator backend). For example, in the keyed 
> backend hierarchy, {{AbstractKeyedStateBackend}} has caches that shortcuts 
> re-accessing already registered state. This behaviour is missing in the 
> operator backend hierarchy, and for example needs to be explicitly handled by 
> the concrete {{DefaultOperatorStateBackend}} subclass implementation.
> As of now, the need of a proper hierarchy also on the operator backend side 
> might not be that prominent, but will mostly likely become more prominent  as 
> we wish to introduce more state structures for operator state (e.g. a 
> {{MapState}} for operator state has already been discussed a few times 
> already) as well as more options besides memory-backed operator state.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Comment Edited] (FLINK-6849) Refactor operator state backend and internal operator state hierarchy

[Stefan Richter (JIRA)]

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

Stefan Richter edited comment on FLINK-6849 at 6/7/17 8:09 AM:
---

I agree in general, because this was clear from the beginning that we need 
might have other implementations of the {{OperatorStateBackend}} interface. 
However, I think we should only start doing this change when we already have a 
concrete plan to introduce more implementations, e.g. based on RocksDB. I would 
also suggest that before starting this, we also spend some thoughts on the 
relationship of operator state and keyed state in general and if they could not 
be based on one common backend abstraction. For now, is there already any 
concrete need to do changes (other than cosmetical)? 


was (Author: srichter):
I agree in general, because this was clear from the beginning that we need 
might have other implementations of the `OperatorStateBackend` interface. 
However, I think we should only start doing this change when we already have a 
concrete plan to introduce more implementations, e.g. based on RocksDB. I would 
also suggest that before starting this, we also spend some thoughts on the 
relationship of operator state and keyed state in general and if they could not 
be based on one common backend abstraction. For now, is there already any 
concrete need to do changes (other than cosmetical)? 

> Refactor operator state backend and internal operator state hierarchy
> -
>
> Key: FLINK-6849
> URL: https://issues.apache.org/jira/browse/FLINK-6849
> Project: Flink
>  Issue Type: Improvement
>  Components: State Backends, Checkpointing
>Reporter: Tzu-Li (Gordon) Tai
>
> Currently, compared to the keyed state backends, the operator state backends, 
> as well as operator state interfaces, lacks proper hierarchy.
> One issue with this lack of hierarchy is that the general concerns of 
> implementing state registration is different between the keyed and operator 
> backends (aside from what is naturally different, such as namespace and key 
> which is not relevant for the operator backend). For example, in the keyed 
> backend hierarchy, {{AbstractKeyedStateBackend}} has caches that shortcuts 
> re-accessing already registered state. This behaviour is missing in the 
> operator backend hierarchy, and for example needs to be explicitly handled by 
> the concrete {{DefaultOperatorStateBackend}} subclass implementation.
> As of now, the need of a proper hierarchy also on the operator backend side 
> might not be that prominent, but will mostly likely become more prominent  as 
> we wish to introduce more state structures for operator state (e.g. a 
> {{MapState}} for operator state has already been discussed a few times 
> already) as well as more options besides memory-backed operator state.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Created] (FLINK-6861) Use OperatorID in metric system

[Chesnay Schepler (JIRA)]

Chesnay Schepler created FLINK-6861:
---

 Summary: Use OperatorID in metric system
 Key: FLINK-6861
 URL: https://issues.apache.org/jira/browse/FLINK-6861
 Project: Flink
  Issue Type: Improvement
  Components: Metrics
Affects Versions: 1.4.0
Reporter: Chesnay Schepler
Assignee: Chesnay Schepler
 Fix For: 1.4.0


The metric system currently identifies operators by name, which frequently 
leads to problems when operators in chain have the same name.

We recently introduced actual operator IDs into the runtime. We should look 
into whether we can expose these easily to the metric system.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6281) Create TableSink for JDBC

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6281:
---

Github user zentol commented on a diff in the pull request:

https://github.com/apache/flink/pull/3712#discussion_r120576032
  
--- Diff: 
flink-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JDBCTableSinkTest.java
 ---
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.java.io.jdbc;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotSame;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Test for JDBCTableSink.
+ */
+public class JDBCTableSinkTest {
+   private static final String[] FIELD_NAMES = new String[]{"foo"};
+   private static final TypeInformation[] FIELD_TYPES = new 
TypeInformation[]{
+   BasicTypeInfo.STRING_TYPE_INFO
+   };
+
+   @Test
+   public void testOutputSink() throws Exception {
+   JDBCOutputFormat outputFormat = mock(JDBCOutputFormat.class);
+   JDBCTableSink sink = new JDBCTableSink(outputFormat);
+   @SuppressWarnings("unchecked")
+   DataStream dataStream = (DataStream) 
mock(DataStream.class);
+   sink.emitDataStream(dataStream);
+   verify(dataStream).addSink(sink);
--- End diff --

you don't have to test this, as it is not a detail of the JDBCTableSink but 
the table API.


> Create TableSink for JDBC
> -
>
> Key: FLINK-6281
> URL: https://issues.apache.org/jira/browse/FLINK-6281
> Project: Flink
>  Issue Type: Improvement
>  Components: Table API & SQL
>Reporter: Haohui Mai
>Assignee: Haohui Mai
>
> It would be nice to integrate the table APIs with the JDBC connectors so that 
> the rows in the tables can be directly pushed into JDBC.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink pull request #3712: [FLINK-6281] Create TableSink for JDBC.

[zentol]

Github user zentol commented on a diff in the pull request:

https://github.com/apache/flink/pull/3712#discussion_r120576118
  
--- Diff: 
flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCOutputFormat.java
 ---
@@ -202,14 +202,20 @@ public void writeRecord(Row row) throws IOException {
upload.addBatch();
batchCount++;
if (batchCount >= batchInterval) {
-   upload.executeBatch();
-   batchCount = 0;
+   flush();
}
} catch (SQLException | IllegalArgumentException e) {
throw new IllegalArgumentException("writeRecord() 
failed", e);
}
}
 
+   void flush() throws SQLException {
+   if (upload != null) {
+   upload.executeBatch();
--- End diff --

It's been a while since i worked with JDBC, I take it this is a synchronous 
call? What happens if this call fails?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6849) Refactor operator state backend and internal operator state hierarchy

[Tzu-Li (Gordon) Tai (JIRA)]

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

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


Hi [~mingleizhang], your proposals also match what I had in mind. Basically, we 
should have an abstract base class for the operator backends (although there is 
only a single {{DefaultOperatorStateBackend}} implementation at the moment).

For this issue, before jumping in, I would also wait a bit for others who are 
more knowledgable on the original design choice there for the state backends 
hierarchy (cc [~aljoscha] [~srichter]) to also have some consensus on the 
change.

> Refactor operator state backend and internal operator state hierarchy
> -
>
> Key: FLINK-6849
> URL: https://issues.apache.org/jira/browse/FLINK-6849
> Project: Flink
>  Issue Type: Improvement
>  Components: State Backends, Checkpointing
>Reporter: Tzu-Li (Gordon) Tai
>
> Currently, compared to the keyed state backends, the operator state backends, 
> as well as operator state interfaces, lacks proper hierarchy.
> One issue with this lack of hierarchy is that the general concerns of 
> implementing state registration is different between the keyed and operator 
> backends (aside from what is naturally different, such as namespace and key 
> which is not relevant for the operator backend). For example, in the keyed 
> backend hierarchy, {{AbstractKeyedStateBackend}} has caches that shortcuts 
> re-accessing already registered state. This behaviour is missing in the 
> operator backend hierarchy, and for example needs to be explicitly handled by 
> the concrete {{DefaultOperatorStateBackend}} subclass implementation.
> As of now, the need of a proper hierarchy also on the operator backend side 
> might not be that prominent, but will mostly likely become more prominent  as 
> we wish to introduce more state structures for operator state (e.g. a 
> {{MapState}} for operator state has already been discussed a few times 
> already) as well as more options besides memory-backed operator state.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Closed] (FLINK-5053) Incremental / lightweight snapshots for checkpoints

[Stefan Richter (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-5053?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Stefan Richter closed FLINK-5053.
-
Resolution: Implemented

> Incremental / lightweight snapshots for checkpoints
> ---
>
> Key: FLINK-5053
> URL: https://issues.apache.org/jira/browse/FLINK-5053
> Project: Flink
>  Issue Type: Improvement
>  Components: State Backends, Checkpointing
>Reporter: Stefan Richter
>Assignee: Xiaogang Shi
>
> There is currently basically no difference between savepoints and checkpoints 
> in Flink and both are created through exactly the same process.
> However, savepoints and checkpoints have a slightly different meaning which 
> we should take into account to keep Flink efficient:
> - Savepoints are (typically infrequently) triggered by the user to create a 
> state from which the application can be restarted, e.g. because Flink, some 
> code, or the parallelism needs to be changed.
> - Checkpoints are (typically frequently) triggered by the System to allow for 
> fast recovery in case of failure, but keeping the job/system unchanged.
> This means that savepoints and checkpoints can have different properties in 
> that:
> - Savepoint should represent a state of the application, where 
> characteristics of the job (e.g. parallelism) can be adjusted for the next 
> restart. One example for things that savepoints need to be aware of are 
> key-groups. Savepoints can potentially be a little more expensive than 
> checkpoints, because they are usually created a lot less frequently through 
> the user.
> - Checkpoints are frequently triggered by the system to allow for fast 
> failure recovery. However, failure recovery leaves all characteristics of the 
> job unchanged. This checkpoints do not have to be aware of those, e.g. think 
> again of key groups. Checkpoints should run faster than creating savepoints, 
> in particular it would be nice to have incremental checkpoints.
> For a first approach, I would suggest the following steps/changes:
> - In checkpoint coordination: differentiate between triggering checkpoints 
> and savepoints. Introduce properties for checkpoints that describe their set 
> of abilities, e.g. "is-key-group-aware", "is-incremental".
> - In state handle infrastructure: introduce state handles that reflect 
> incremental checkpoints and drop full key-group awareness, i.e. covering 
> folders instead of files and not having keygroup_id -> file/offset mapping, 
> but keygroup_range -> folder?
> - Backend side: We should start with RocksDB by reintroducing something 
> similar to semi-async snapshots, but using 
> BackupableDBOptions::setShareTableFiles(true) and transferring only new 
> incremental outputs to HDFS. Notice that using RocksDB's internal backup 
> mechanism is giving up on the information about individual key-groups. But as 
> explained above, this should be totally acceptable for checkpoints, while 
> savepoints should use the key-group-aware fully async mode. Of course we also 
> need to implement the ability to restore from both types of snapshots.
> One problem in the suggested approach is still that even checkpoints should 
> support scale-down, in case that only a smaller number of instances is left 
> available in a recovery case.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Closed] (FLINK-6364) Implement incremental checkpointing in RocksDBStateBackend

[Stefan Richter (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6364?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Stefan Richter closed FLINK-6364.
-
Resolution: Fixed

> Implement incremental checkpointing in RocksDBStateBackend
> --
>
> Key: FLINK-6364
> URL: https://issues.apache.org/jira/browse/FLINK-6364
> Project: Flink
>  Issue Type: Sub-task
>  Components: State Backends, Checkpointing
>Reporter: Xiaogang Shi
>Assignee: Xiaogang Shi
>
> {{RocksDBStateBackend}} is well suited for incremental checkpointing because 
> RocksDB is base on LSM trees,  which record updates in new sst files and all 
> sst files are immutable. By only materializing those new sst files, we can 
> significantly improve the performance of checkpointing.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Comment Edited] (FLINK-6488) Remove 'start-local.sh' script

[mingleizhang (JIRA)]

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

mingleizhang edited comment on FLINK-6488 at 6/7/17 9:25 AM:
-

That makes sense better. Patch will available soon and I would suggest the 
title for this issue should change in a way.


was (Author: mingleizhang):
That makes sense better. Patch will available soon.

> Remove 'start-local.sh' script
> --
>
> Key: FLINK-6488
> URL: https://issues.apache.org/jira/browse/FLINK-6488
> Project: Flink
>  Issue Type: Sub-task
>  Components: Startup Shell Scripts
>Reporter: Stephan Ewen
>Assignee: mingleizhang
>
> The {{start-cluster.sh}} scripts work locally now, without needing SSH setup.
> We can remove {{start-local.sh}} without any loss of functionality.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (FLINK-6866) ClosureCleaner.clean fails for scala's JavaConverters wrapper classes

[SmedbergM (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6866?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

SmedbergM updated FLINK-6866:
-
Description: 
MWE:

```
import scala.collection.JavaConverters._
import org.apache.flink.api.java.ClosureCleaner

object SerializationFailureMWE extends App {
  val m4j: java.util.Map[String,String] = new java.util.HashMap
  m4j.put("key1", "value1")

  val m: java.util.Map[String,String] = Map(
"key1" -> "value1"
  ).asJava

  println("Cleaning native Java map")
  ClosureCleaner.clean(m4j, true)

  println("Cleaning map converted by JavaConverters")
  ClosureCleaner.clean(m, true)
}
```

Program output:
```
Cleaning native Java map
Cleaning map converted by JavaConverters
Exception in thread "main" org.apache.flink.api.common.InvalidProgramException: 
{key1=value1} is not serializable. The object probably contains or references 
non serializable fields.
at 
org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:100)
at 
SerializationFailureMWE$delayedInit$body.apply(SerializationFailureMWE.scala:17)
at scala.Function0$class.apply$mcV$sp(Function0.scala:40)
at 
scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12)
at scala.App$$anonfun$main$1.apply(App.scala:71)
at scala.App$$anonfun$main$1.apply(App.scala:71)
at scala.collection.immutable.List.foreach(List.scala:318)
at 
scala.collection.generic.TraversableForwarder$class.foreach(TraversableForwarder.scala:32)
at scala.App$class.main(App.scala:71)
at SerializationFailureMWE$.main(SerializationFailureMWE.scala:5)
at SerializationFailureMWE.main(SerializationFailureMWE.scala)
Caused by: java.io.NotSerializableException: 
scala.collection.convert.Wrappers$MapWrapper
...
```

  was:
MWE:

```
import scala.collection.JavaConverters._
import org.apache.flink.api.java.ClosureCleaner

object SerializationFailureMWE extends App {
  val m4j: java.util.Map[String,String] = new java.util.HashMap
  m4j.put("key1", "value1")

  val m: java.util.Map[String,String] = Map(
"key1" -> "value1"
  ).asJava

  println("Cleaning native Java map")

  ClosureCleaner.clean(m4j, true)

  println("Cleaning map converted by JavaConverters")

  ClosureCleaner.clean(m, true)
}
```

Program output:
```
Cleaning native Java map
Cleaning map converted by JavaConverters
Exception in thread "main" org.apache.flink.api.common.InvalidProgramException: 
{key1=value1} is not serializable. The object probably contains or references 
non serializable fields.
at 
org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:100)
at 
SerializationFailureMWE$delayedInit$body.apply(SerializationFailureMWE.scala:17)
at scala.Function0$class.apply$mcV$sp(Function0.scala:40)
at 
scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12)
at scala.App$$anonfun$main$1.apply(App.scala:71)
at scala.App$$anonfun$main$1.apply(App.scala:71)
at scala.collection.immutable.List.foreach(List.scala:318)
at 
scala.collection.generic.TraversableForwarder$class.foreach(TraversableForwarder.scala:32)
at scala.App$class.main(App.scala:71)
at SerializationFailureMWE$.main(SerializationFailureMWE.scala:5)
at SerializationFailureMWE.main(SerializationFailureMWE.scala)
Caused by: java.io.NotSerializableException: 
scala.collection.convert.Wrappers$MapWrapper
...
```


> ClosureCleaner.clean fails for scala's JavaConverters wrapper classes
> -
>
> Key: FLINK-6866
> URL: https://issues.apache.org/jira/browse/FLINK-6866
> Project: Flink
>  Issue Type: Bug
>  Components: DataStream API
>Affects Versions: 1.2.0
>Reporter: SmedbergM
>
> MWE:
> ```
> import scala.collection.JavaConverters._
> import org.apache.flink.api.java.ClosureCleaner
> object SerializationFailureMWE extends App {
>   val m4j: java.util.Map[String,String] = new java.util.HashMap
>   m4j.put("key1", "value1")
>   val m: java.util.Map[String,String] = Map(
> "key1" -> "value1"
>   ).asJava
>   println("Cleaning native Java map")
>   ClosureCleaner.clean(m4j, true)
>   println("Cleaning map converted by JavaConverters")
>   ClosureCleaner.clean(m, true)
> }
> ```
> Program output:
> ```
> Cleaning native Java map
> Cleaning map converted by JavaConverters
> Exception in thread "main" 
> org.apache.flink.api.common.InvalidProgramException: {key1=value1} is not 
> serializable. The object probably contains or references non serializable 
> fields.
>   at 
> org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:100)
>   at 
> SerializationFailureMWE$delayedInit$body.apply(SerializationFailureMWE.scala:17)
>   at scala.Function0$class.apply$mcV$sp(Function0.scala:40)
>   at 

[jira] [Updated] (FLINK-6866) ClosureCleaner.clean fails for scala's JavaConverters wrapper classes

[SmedbergM (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6866?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

SmedbergM updated FLINK-6866:
-
Description: 
MWE:

```
import scala.collection.JavaConverters._
import org.apache.flink.api.java.ClosureCleaner

object SerializationFailureMWE extends App {
  val m4j: java.util.Map[String,String] = new java.util.HashMap
  m4j.put("key1", "value1")

  val m: java.util.Map[String,String] = Map(
"key1" -> "value1"
  ).asJava

  println("Cleaning native Java map")

  ClosureCleaner.clean(m4j, true)

  println("Cleaning map converted by JavaConverters")

  ClosureCleaner.clean(m, true)
}
```

Program output:
```
Cleaning native Java map
Cleaning map converted by JavaConverters
Exception in thread "main" org.apache.flink.api.common.InvalidProgramException: 
{key1=value1} is not serializable. The object probably contains or references 
non serializable fields.
at 
org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:100)
at 
SerializationFailureMWE$delayedInit$body.apply(SerializationFailureMWE.scala:17)
at scala.Function0$class.apply$mcV$sp(Function0.scala:40)
at 
scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12)
at scala.App$$anonfun$main$1.apply(App.scala:71)
at scala.App$$anonfun$main$1.apply(App.scala:71)
at scala.collection.immutable.List.foreach(List.scala:318)
at 
scala.collection.generic.TraversableForwarder$class.foreach(TraversableForwarder.scala:32)
at scala.App$class.main(App.scala:71)
at SerializationFailureMWE$.main(SerializationFailureMWE.scala:5)
at SerializationFailureMWE.main(SerializationFailureMWE.scala)
Caused by: java.io.NotSerializableException: 
scala.collection.convert.Wrappers$MapWrapper
...
```

  was:
MWE:

```
import scala.collection.JavaConverters._
import org.apache.flink.api.java.ClosureCleaner

object SerializationFailureMWE extends App {
  val m4j: java.util.Map[String,String] = new java.util.HashMap
  m4j.put("key1", "value1")

  val m: java.util.Map[String,String] = Map(
"key1" -> "value1"
  ).asJava

  println("Cleaning native Java map")
  ClosureCleaner.clean(m4j, true)

  println("Cleaning map converted by JavaConverters")
  ClosureCleaner.clean(m, true)
}
```

Program output:
```
Cleaning native Java map
Cleaning map converted by JavaConverters
Exception in thread "main" org.apache.flink.api.common.InvalidProgramException: 
{key1=value1} is not serializable. The object probably contains or references 
non serializable fields.
at 
org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:100)
at 
SerializationFailureMWE$delayedInit$body.apply(SerializationFailureMWE.scala:17)
at scala.Function0$class.apply$mcV$sp(Function0.scala:40)
at 
scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12)
at scala.App$$anonfun$main$1.apply(App.scala:71)
at scala.App$$anonfun$main$1.apply(App.scala:71)
at scala.collection.immutable.List.foreach(List.scala:318)
at 
scala.collection.generic.TraversableForwarder$class.foreach(TraversableForwarder.scala:32)
at scala.App$class.main(App.scala:71)
at SerializationFailureMWE$.main(SerializationFailureMWE.scala:5)
at SerializationFailureMWE.main(SerializationFailureMWE.scala)
Caused by: java.io.NotSerializableException: 
scala.collection.convert.Wrappers$MapWrapper
...
```


> ClosureCleaner.clean fails for scala's JavaConverters wrapper classes
> -
>
> Key: FLINK-6866
> URL: https://issues.apache.org/jira/browse/FLINK-6866
> Project: Flink
>  Issue Type: Bug
>  Components: DataStream API
>Affects Versions: 1.2.0
>Reporter: SmedbergM
>
> MWE:
> ```
> import scala.collection.JavaConverters._
> import org.apache.flink.api.java.ClosureCleaner
> object SerializationFailureMWE extends App {
>   val m4j: java.util.Map[String,String] = new java.util.HashMap
>   m4j.put("key1", "value1")
>   val m: java.util.Map[String,String] = Map(
> "key1" -> "value1"
>   ).asJava
>   println("Cleaning native Java map")
>   ClosureCleaner.clean(m4j, true)
>   println("Cleaning map converted by JavaConverters")
>   ClosureCleaner.clean(m, true)
> }
> ```
> Program output:
> ```
> Cleaning native Java map
> Cleaning map converted by JavaConverters
> Exception in thread "main" 
> org.apache.flink.api.common.InvalidProgramException: {key1=value1} is not 
> serializable. The object probably contains or references non serializable 
> fields.
>   at 
> org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:100)
>   at 
> SerializationFailureMWE$delayedInit$body.apply(SerializationFailureMWE.scala:17)
>   at scala.Function0$class.apply$mcV$sp(Function0.scala:40)
>   at 

[jira] [Commented] (FLINK-6865) Expand checkstyle docs to include import in intellij

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6865:
---

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4086
  
Here's a list of the currently supported modules btw.:
```
EmptyLineSeparator
FileTabCharacter
Indentation
LeftCurly
LineLength
NeedBraces
NoWhitespacesBefore
WhitespaceAfter
WhitespaceAround
```


> Expand checkstyle docs to include import in intellij
> 
>
> Key: FLINK-6865
> URL: https://issues.apache.org/jira/browse/FLINK-6865
> Project: Flink
>  Issue Type: Improvement
>  Components: Checkstyle
>Affects Versions: 1.4.0
>Reporter: Chesnay Schepler
>Assignee: Chesnay Schepler
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Created] (FLINK-6866) ClosureCleaner.clean fails for scala's JavaConverters wrapper classes

[SmedbergM (JIRA)]

SmedbergM created FLINK-6866:


 Summary: ClosureCleaner.clean fails for scala's JavaConverters 
wrapper classes
 Key: FLINK-6866
 URL: https://issues.apache.org/jira/browse/FLINK-6866
 Project: Flink
  Issue Type: Bug
  Components: DataStream API
Affects Versions: 1.2.0
Reporter: SmedbergM


MWE:

```
import scala.collection.JavaConverters._
import org.apache.flink.api.java.ClosureCleaner

object SerializationFailureMWE extends App {
  val m4j: java.util.Map[String,String] = new java.util.HashMap
  m4j.put("key1", "value1")

  val m: java.util.Map[String,String] = Map(
"key1" -> "value1"
  ).asJava

  println("Cleaning native Java map")
  ClosureCleaner.clean(m4j, true)

  println("Cleaning map converted by JavaConverters")
  ClosureCleaner.clean(m, true)
}
```

Program output:
```
Cleaning native Java map
Cleaning map converted by JavaConverters
Exception in thread "main" org.apache.flink.api.common.InvalidProgramException: 
{key1=value1} is not serializable. The object probably contains or references 
non serializable fields.
at 
org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:100)
at 
SerializationFailureMWE$delayedInit$body.apply(SerializationFailureMWE.scala:17)
at scala.Function0$class.apply$mcV$sp(Function0.scala:40)
at 
scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12)
at scala.App$$anonfun$main$1.apply(App.scala:71)
at scala.App$$anonfun$main$1.apply(App.scala:71)
at scala.collection.immutable.List.foreach(List.scala:318)
at 
scala.collection.generic.TraversableForwarder$class.foreach(TraversableForwarder.scala:32)
at scala.App$class.main(App.scala:71)
at SerializationFailureMWE$.main(SerializationFailureMWE.scala:5)
at SerializationFailureMWE.main(SerializationFailureMWE.scala)
Caused by: java.io.NotSerializableException: 
scala.collection.convert.Wrappers$MapWrapper
...
```



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink issue #4086: [FLINK-6865] Update checkstyle documentation

[zentol]

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4086
  
Only a (small) subset of checkstyle modules can be moduled, which sadly 
doesn't include the `ImportOrder` module.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6865) Expand checkstyle docs to include import in intellij

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6865:
---

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4086
  
Only a (small) subset of checkstyle modules can be moduled, which sadly 
doesn't include the `ImportOrder` module.


> Expand checkstyle docs to include import in intellij
> 
>
> Key: FLINK-6865
> URL: https://issues.apache.org/jira/browse/FLINK-6865
> Project: Flink
>  Issue Type: Improvement
>  Components: Checkstyle
>Affects Versions: 1.4.0
>Reporter: Chesnay Schepler
>Assignee: Chesnay Schepler
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink issue #4086: [FLINK-6865] Update checkstyle documentation

[zentol]

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4086
  
Here's a list of the currently supported modules btw.:
```
EmptyLineSeparator
FileTabCharacter
Indentation
LeftCurly
LineLength
NeedBraces
NoWhitespacesBefore
WhitespaceAfter
WhitespaceAround
```


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6865) Expand checkstyle docs to include import in intellij

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6865:
---

Github user greghogan commented on the issue:

https://github.com/apache/flink/pull/4086
  
Okay, second best may be to create an IntelliJ Code Style configuration for 
developers to import.


> Expand checkstyle docs to include import in intellij
> 
>
> Key: FLINK-6865
> URL: https://issues.apache.org/jira/browse/FLINK-6865
> Project: Flink
>  Issue Type: Improvement
>  Components: Checkstyle
>Affects Versions: 1.4.0
>Reporter: Chesnay Schepler
>Assignee: Chesnay Schepler
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink issue #4086: [FLINK-6865] Update checkstyle documentation

[greghogan]

Github user greghogan commented on the issue:

https://github.com/apache/flink/pull/4086
  
Okay, second best may be to create an IntelliJ Code Style configuration for 
developers to import.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6865) Expand checkstyle docs to include import in intellij

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6865:
---

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4086
  
I don't know a lot about the intellij code style config; can we only define 
subset that is imported without affecting the rest?

Btw., I'm looking into contributing to the checkstyle plugin to add support 
for the rules that we need.


> Expand checkstyle docs to include import in intellij
> 
>
> Key: FLINK-6865
> URL: https://issues.apache.org/jira/browse/FLINK-6865
> Project: Flink
>  Issue Type: Improvement
>  Components: Checkstyle
>Affects Versions: 1.4.0
>Reporter: Chesnay Schepler
>Assignee: Chesnay Schepler
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink issue #4086: [FLINK-6865] Update checkstyle documentation

[zentol]

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4086
  
I don't know a lot about the intellij code style config; can we only define 
subset that is imported without affecting the rest?

Btw., I'm looking into contributing to the checkstyle plugin to add support 
for the rules that we need.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Created] (FLINK-6862) Tumble window rowtime not resolve at logic plan validation

[Mark You (JIRA)]

Mark You created FLINK-6862:
---

 Summary: Tumble window rowtime not resolve at logic plan validation
 Key: FLINK-6862
 URL: https://issues.apache.org/jira/browse/FLINK-6862
 Project: Flink
  Issue Type: Bug
  Components: Table API & SQL
Affects Versions: 1.3.0
Reporter: Mark You


Following code sample work in version 1.2.1, but failed at 1.3.0
{code:title=Bar.java|borderStyle=solid}
public class TumblingWindow {

public static void main(String[] args) throws Exception {
List data = new ArrayList();
data.add(new Content(1L, "Hi"));
data.add(new Content(2L, "Hallo"));
data.add(new Content(3L, "Hello"));
data.add(new Content(4L, "Hello"));
data.add(new Content(7L, "Hello"));
data.add(new Content(8L, "Hello world"));
data.add(new Content(16L, "Hello world"));

final StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

final StreamTableEnvironment tableEnv = 
TableEnvironment.getTableEnvironment(env);

DataStream stream = env.fromCollection(data);

DataStream stream2 = stream.assignTimestampsAndWatermarks(
new 
BoundedOutOfOrdernessTimestampExtractor(Time.milliseconds(1)) {

/**
 * 
 */
private static final long serialVersionUID = 
410512296011057717L;

@Override
public long extractTimestamp(Content element) {
return element.getRecordTime();
}

});

Table table = tableEnv.fromDataStream(stream2);

table.window(Tumble.over("1.hours").on("rowtime").as("w")).groupBy("w").select("w.start,
 content.count");

env.execute();
}

public static class Content implements Serializable {

private long recordTime;
private String content;

public Content() {
super();
}

public Content(long recordTime, String content) {
super();
this.recordTime = recordTime;
this.content = content;
}

public long getRecordTime() {
return recordTime;
}

public void setRecordTime(long recordTime) {
this.recordTime = recordTime;
}

public String getContent() {
return content;
}

public void setContent(String content) {
this.content = content;
}

}

private class TimestampWithEqualWatermark implements 
AssignerWithPunctuatedWatermarks {

/**
 * 
 */
private static final long serialVersionUID = 1L;

@Override
public long extractTimestamp(Object[] element, long 
previousElementTimestamp) {
// TODO Auto-generated method stub
return (long) element[0];
}

@Override
public Watermark checkAndGetNextWatermark(Object[] lastElement, long 
extractedTimestamp) {
return new Watermark(extractedTimestamp);
}

}
}
{code}

{noformat}
Exception trace
Exception in thread "main" org.apache.flink.table.api.ValidationException: 
Cannot resolve [rowtime] given input [content, recordTime].
at 
org.apache.flink.table.plan.logical.LogicalNode.failValidation(LogicalNode.scala:143)
at 
org.apache.flink.table.plan.logical.LogicalNode$$anonfun$validate$1.applyOrElse(LogicalNode.scala:86)
at 
org.apache.flink.table.plan.logical.LogicalNode$$anonfun$validate$1.applyOrElse(LogicalNode.scala:83)
at 
org.apache.flink.table.plan.TreeNode.postOrderTransform(TreeNode.scala:72)
at 
org.apache.flink.table.plan.logical.LogicalNode.org$apache$flink$table$plan$logical$LogicalNode$$expressionPostOrderTransform$1(LogicalNode.scala:119)
at 
org.apache.flink.table.plan.logical.LogicalNode$$anonfun$7$$anonfun$apply$1.apply(LogicalNode.scala:132)
at 
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
at 
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
at 
scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:245)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at 
org.apache.flink.table.plan.logical.LogicalNode$$anonfun$7.apply(LogicalNode.scala:131)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:370)
at scala.collection.Iterator$class.foreach(Iterator.scala:742)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1194)

[jira] [Updated] (FLINK-6862) Tumble window rowtime not resolve at logic plan validation

[Mark You (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6862?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Mark You updated FLINK-6862:

Description: 
Following code sample work in version 1.2.1, but failed at 1.3.0
{code:title= TumblingWindow.java|borderStyle=solid}
public class TumblingWindow {

public static void main(String[] args) throws Exception {
List data = new ArrayList();
data.add(new Content(1L, "Hi"));
data.add(new Content(2L, "Hallo"));
data.add(new Content(3L, "Hello"));
data.add(new Content(4L, "Hello"));
data.add(new Content(7L, "Hello"));
data.add(new Content(8L, "Hello world"));
data.add(new Content(16L, "Hello world"));

final StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

final StreamTableEnvironment tableEnv = 
TableEnvironment.getTableEnvironment(env);

DataStream stream = env.fromCollection(data);

DataStream stream2 = stream.assignTimestampsAndWatermarks(
new 
BoundedOutOfOrdernessTimestampExtractor(Time.milliseconds(1)) {

/**
 * 
 */
private static final long serialVersionUID = 
410512296011057717L;

@Override
public long extractTimestamp(Content element) {
return element.getRecordTime();
}

});

Table table = tableEnv.fromDataStream(stream2);

table.window(Tumble.over("1.hours").on("rowtime").as("w")).groupBy("w").select("w.start,
 content.count");

env.execute();
}

public static class Content implements Serializable {

private long recordTime;
private String content;

public Content() {
super();
}

public Content(long recordTime, String content) {
super();
this.recordTime = recordTime;
this.content = content;
}

public long getRecordTime() {
return recordTime;
}

public void setRecordTime(long recordTime) {
this.recordTime = recordTime;
}

public String getContent() {
return content;
}

public void setContent(String content) {
this.content = content;
}

}

private class TimestampWithEqualWatermark implements 
AssignerWithPunctuatedWatermarks {

/**
 * 
 */
private static final long serialVersionUID = 1L;

@Override
public long extractTimestamp(Object[] element, long 
previousElementTimestamp) {
// TODO Auto-generated method stub
return (long) element[0];
}

@Override
public Watermark checkAndGetNextWatermark(Object[] lastElement, long 
extractedTimestamp) {
return new Watermark(extractedTimestamp);
}

}
}
{code}

{noformat}
Exception trace
Exception in thread "main" org.apache.flink.table.api.ValidationException: 
Cannot resolve [rowtime] given input [content, recordTime].
at 
org.apache.flink.table.plan.logical.LogicalNode.failValidation(LogicalNode.scala:143)
at 
org.apache.flink.table.plan.logical.LogicalNode$$anonfun$validate$1.applyOrElse(LogicalNode.scala:86)
at 
org.apache.flink.table.plan.logical.LogicalNode$$anonfun$validate$1.applyOrElse(LogicalNode.scala:83)
at 
org.apache.flink.table.plan.TreeNode.postOrderTransform(TreeNode.scala:72)
at 
org.apache.flink.table.plan.logical.LogicalNode.org$apache$flink$table$plan$logical$LogicalNode$$expressionPostOrderTransform$1(LogicalNode.scala:119)
at 
org.apache.flink.table.plan.logical.LogicalNode$$anonfun$7$$anonfun$apply$1.apply(LogicalNode.scala:132)
at 
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
at 
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
at 
scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:245)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at 
org.apache.flink.table.plan.logical.LogicalNode$$anonfun$7.apply(LogicalNode.scala:131)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:370)
at scala.collection.Iterator$class.foreach(Iterator.scala:742)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1194)
at 
scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
at 
scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
at 

[GitHub] flink pull request #4048: [FLINK-6812] Enforce Java8 when creating a release

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/4048


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6844) TraversableSerializer should implement compatibility methods

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6844:
---

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/4081
  
Tested this with a stateful job that uses Scala collections as state.
Merging this ..


> TraversableSerializer should implement compatibility methods
> 
>
> Key: FLINK-6844
> URL: https://issues.apache.org/jira/browse/FLINK-6844
> Project: Flink
>  Issue Type: Bug
>  Components: Type Serialization System
>Affects Versions: 1.3.0
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
>Priority: Blocker
>  Labels: flink-rel-1.3.1-blockers
> Fix For: 1.3.1
>
>
> The {{TraversableSerializer}} may be used as a serializer for managed state 
> and takes part in checkpointing, therefore should implement the compatibility 
> methods.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6661) Merge "Subtasks" and "TaskManagers" view

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6661:
---

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/3998


> Merge "Subtasks" and "TaskManagers" view
> 
>
> Key: FLINK-6661
> URL: https://issues.apache.org/jira/browse/FLINK-6661
> Project: Flink
>  Issue Type: Sub-task
>  Components: Webfrontend
>Affects Versions: 1.4.0
>Reporter: Chesnay Schepler
>Assignee: Chesnay Schepler
> Fix For: 1.4.0
>
>
> The {{Subtasks}} and {{TaskManagers/ Subtasks by TaskManager}} view are very 
> similar in what they do, so much in fact that they are identical if the 
> taskmanagers have a single task slot.
> I propose to merge them, and add a checkbox to aggregate the subtasks by 
> taskmanagers.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink issue #4069: [FLINK-6723] Activate strict checkstyle for flink-librari...

[zentol]

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4069
  
merging.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6723) Activate strict checkstyle for flink-libraries

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6723:
---

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4069
  
merging.


> Activate strict checkstyle for flink-libraries
> --
>
> Key: FLINK-6723
> URL: https://issues.apache.org/jira/browse/FLINK-6723
> Project: Flink
>  Issue Type: Sub-task
>  Components: CEP, Gelly, Machine Learning Library, Python API, Table 
> API & SQL
>Reporter: Chesnay Schepler
>Assignee: Chesnay Schepler
>Priority: Trivial
> Fix For: 1.4.0
>
>
> Move checkstyle plugin into flink-libraries pom once the following issues are 
> resolved:
> FLINK-6137
> FLINK-6432
> FLINK-6709
> FLINK-6722



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink issue #4039: [FLINK-6783] Changed passing index of type argument while...

[dawidwys]

Github user dawidwys commented on the issue:

https://github.com/apache/flink/pull/4039
  
I will address them today. I am working on last comment with enabling the 
changes also for Partitioner.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6783) Wrongly extracted TypeInformations for WindowedStream::aggregate

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6783:
---

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/4039
  
Cool, thank you!


> Wrongly extracted TypeInformations for WindowedStream::aggregate
> 
>
> Key: FLINK-6783
> URL: https://issues.apache.org/jira/browse/FLINK-6783
> Project: Flink
>  Issue Type: Bug
>  Components: Core, DataStream API
>Affects Versions: 1.3.0, 1.3.1
>Reporter: Dawid Wysakowicz
>Assignee: Dawid Wysakowicz
>Priority: Blocker
>  Labels: flink-rel-1.3.1-blockers
> Fix For: 1.3.1
>
>
> The following test fails because of wrongly acquired output type for 
> {{AggregateFunction}}:
> {code}
> @Test
> public void testAggregateWithWindowFunctionDifferentResultTypes() throws 
> Exception {
>   StreamExecutionEnvironment env = 
> StreamExecutionEnvironment.getExecutionEnvironment();
>   DataStream> source = 
> env.fromElements(Tuple2.of("hello", 1), Tuple2.of("hello", 2));
>   DataStream> window = source
>   .keyBy(new TupleKeySelector())
>   .window(TumblingEventTimeWindows.of(Time.of(1, 
> TimeUnit.SECONDS)))
>   .aggregate(new AggregateFunction, 
> Tuple2, String>() {
>   @Override
>   public Tuple2 createAccumulator() {
>   return Tuple2.of("", 0);
>   }
>   @Override
>   public void add(
>   Tuple2 value, Tuple2 Integer> accumulator) {
>   }
>   @Override
>   public String getResult(Tuple2 
> accumulator) {
>   return accumulator.f0;
>   }
>   @Override
>   public Tuple2 merge(
>   Tuple2 a, Tuple2 Integer> b) {
>   return Tuple2.of("", 0);
>   }
>   }, new WindowFunction, 
> String, TimeWindow>() {
>   @Override
>   public void apply(
>   String s,
>   TimeWindow window,
>   Iterable input,
>   Collector> out) 
> throws Exception {
>   out.collect(Tuple3.of("", "", 0));
>   }
>   });
>   OneInputTransformation, Tuple3 Integer>> transform =
>   (OneInputTransformation, Tuple3 String, Integer>>) window.getTransformation();
>   OneInputStreamOperator, Tuple3 Integer>> operator = transform.getOperator();
>   Assert.assertTrue(operator instanceof WindowOperator);
>   WindowOperator, ?, ?, ?> winOperator =
>   (WindowOperator, ?, ?, ?>) 
> operator;
>   Assert.assertTrue(winOperator.getTrigger() instanceof EventTimeTrigger);
>   Assert.assertTrue(winOperator.getWindowAssigner() instanceof 
> TumblingEventTimeWindows);
>   Assert.assertTrue(winOperator.getStateDescriptor() instanceof 
> AggregatingStateDescriptor);
>   processElementAndEnsureOutput(
>   operator, winOperator.getKeySelector(), 
> BasicTypeInfo.STRING_TYPE_INFO, new Tuple2<>("hello", 1));
> }
> {code}
> The test results in 
> {code}
> org.apache.flink.api.common.functions.InvalidTypesException: Input mismatch: 
> Tuple type expected.
>   at 
> org.apache.flink.api.java.typeutils.TypeExtractor.validateInputType(TypeExtractor.java:1157)
>   at 
> org.apache.flink.api.java.typeutils.TypeExtractor.getUnaryOperatorReturnType(TypeExtractor.java:451)
>   at 
> org.apache.flink.streaming.api.datastream.WindowedStream.aggregate(WindowedStream.java:855)
>   at 
> org.apache.flink.streaming.runtime.operators.windowing.WindowTranslationTest.testAggregateWithWindowFunctionDifferentResultTypes(WindowTranslationTest.java:702)
>   at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>   at 
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
>   at 
> 

[GitHub] flink issue #4039: [FLINK-6783] Changed passing index of type argument while...

[rmetzger]

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/4039
  
Cool, thank you!


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6859) StateCleaningCountTrigger should not delete timer

[sunjincheng (JIRA)]

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

sunjincheng commented on FLINK-6859:


Get set of registered timers by namespace. And traverse the timer queue to 
delete the timers. is really expensive operation. I'll remove it .

> StateCleaningCountTrigger should not delete timer
> -
>
> Key: FLINK-6859
> URL: https://issues.apache.org/jira/browse/FLINK-6859
> Project: Flink
>  Issue Type: Improvement
>  Components: Table API & SQL
>Affects Versions: 1.3.0
>Reporter: Fabian Hueske
>
> The {{StateCleaningCountTrigger}} which is used to clean-up inactive state 
> should not delete timers, i.e.. not call {{deleteProcessingTimeTimer()}}.
> This is an expensive operation.
> We should rather fire the timer and check if we need to clean the state or 
> not.
> What do you think [~sunjincheng121]?



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink issue #4048: [FLINK-6812] Enforce Java8 when creating a release

[rmetzger]

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/4048
  
I've tested the change with the release scripts, and the elasticsearch5 
connector was showing up.

I'll merge the change now.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6812) Elasticsearch 5 release artifacts not published to Maven central

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6812:
---

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/4048


> Elasticsearch 5 release artifacts not published to Maven central
> 
>
> Key: FLINK-6812
> URL: https://issues.apache.org/jira/browse/FLINK-6812
> Project: Flink
>  Issue Type: Bug
>  Components: ElasticSearch Connector
>Affects Versions: 1.3.0
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Robert Metzger
>Priority: Blocker
>  Labels: flink-rel-1.3.1-blockers
> Fix For: 1.3.1
>
>
> Release artifacts for the Elasticsearch 5 connector is not published to the 
> Maven Central. Elasticsearch 5 requires Java 8 at minimum, so for the release 
> we need to build with Java 8 for this.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink pull request #3998: [FLINK-6661][web] Merge "Subtasks" and "SubtasksBy...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/3998


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Closed] (FLINK-6661) Merge "Subtasks" and "TaskManagers" view

[Chesnay Schepler (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6661?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Chesnay Schepler closed FLINK-6661.
---
   Resolution: Fixed
Fix Version/s: 1.4.0

1.4: c595502148de7dc02a407480570f9e51df81b268

> Merge "Subtasks" and "TaskManagers" view
> 
>
> Key: FLINK-6661
> URL: https://issues.apache.org/jira/browse/FLINK-6661
> Project: Flink
>  Issue Type: Sub-task
>  Components: Webfrontend
>Affects Versions: 1.4.0
>Reporter: Chesnay Schepler
>Assignee: Chesnay Schepler
> Fix For: 1.4.0
>
>
> The {{Subtasks}} and {{TaskManagers/ Subtasks by TaskManager}} view are very 
> similar in what they do, so much in fact that they are identical if the 
> taskmanagers have a single task slot.
> I propose to merge them, and add a checkbox to aggregate the subtasks by 
> taskmanagers.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6783) Wrongly extracted TypeInformations for WindowedStream::aggregate

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6783:
---

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/4039
  
@dawidwys What's your schedule to address the comments?
This is one of the real blockers for the 1.3.1 release, and I would like to 
put the first RC this week.


> Wrongly extracted TypeInformations for WindowedStream::aggregate
> 
>
> Key: FLINK-6783
> URL: https://issues.apache.org/jira/browse/FLINK-6783
> Project: Flink
>  Issue Type: Bug
>  Components: Core, DataStream API
>Affects Versions: 1.3.0, 1.3.1
>Reporter: Dawid Wysakowicz
>Assignee: Dawid Wysakowicz
>Priority: Blocker
>  Labels: flink-rel-1.3.1-blockers
> Fix For: 1.3.1
>
>
> The following test fails because of wrongly acquired output type for 
> {{AggregateFunction}}:
> {code}
> @Test
> public void testAggregateWithWindowFunctionDifferentResultTypes() throws 
> Exception {
>   StreamExecutionEnvironment env = 
> StreamExecutionEnvironment.getExecutionEnvironment();
>   DataStream> source = 
> env.fromElements(Tuple2.of("hello", 1), Tuple2.of("hello", 2));
>   DataStream> window = source
>   .keyBy(new TupleKeySelector())
>   .window(TumblingEventTimeWindows.of(Time.of(1, 
> TimeUnit.SECONDS)))
>   .aggregate(new AggregateFunction, 
> Tuple2, String>() {
>   @Override
>   public Tuple2 createAccumulator() {
>   return Tuple2.of("", 0);
>   }
>   @Override
>   public void add(
>   Tuple2 value, Tuple2 Integer> accumulator) {
>   }
>   @Override
>   public String getResult(Tuple2 
> accumulator) {
>   return accumulator.f0;
>   }
>   @Override
>   public Tuple2 merge(
>   Tuple2 a, Tuple2 Integer> b) {
>   return Tuple2.of("", 0);
>   }
>   }, new WindowFunction, 
> String, TimeWindow>() {
>   @Override
>   public void apply(
>   String s,
>   TimeWindow window,
>   Iterable input,
>   Collector> out) 
> throws Exception {
>   out.collect(Tuple3.of("", "", 0));
>   }
>   });
>   OneInputTransformation, Tuple3 Integer>> transform =
>   (OneInputTransformation, Tuple3 String, Integer>>) window.getTransformation();
>   OneInputStreamOperator, Tuple3 Integer>> operator = transform.getOperator();
>   Assert.assertTrue(operator instanceof WindowOperator);
>   WindowOperator, ?, ?, ?> winOperator =
>   (WindowOperator, ?, ?, ?>) 
> operator;
>   Assert.assertTrue(winOperator.getTrigger() instanceof EventTimeTrigger);
>   Assert.assertTrue(winOperator.getWindowAssigner() instanceof 
> TumblingEventTimeWindows);
>   Assert.assertTrue(winOperator.getStateDescriptor() instanceof 
> AggregatingStateDescriptor);
>   processElementAndEnsureOutput(
>   operator, winOperator.getKeySelector(), 
> BasicTypeInfo.STRING_TYPE_INFO, new Tuple2<>("hello", 1));
> }
> {code}
> The test results in 
> {code}
> org.apache.flink.api.common.functions.InvalidTypesException: Input mismatch: 
> Tuple type expected.
>   at 
> org.apache.flink.api.java.typeutils.TypeExtractor.validateInputType(TypeExtractor.java:1157)
>   at 
> org.apache.flink.api.java.typeutils.TypeExtractor.getUnaryOperatorReturnType(TypeExtractor.java:451)
>   at 
> org.apache.flink.streaming.api.datastream.WindowedStream.aggregate(WindowedStream.java:855)
>   at 
> org.apache.flink.streaming.runtime.operators.windowing.WindowTranslationTest.testAggregateWithWindowFunctionDifferentResultTypes(WindowTranslationTest.java:702)
>   at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>   at 
> 

[GitHub] flink issue #4039: [FLINK-6783] Changed passing index of type argument while...

[rmetzger]

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/4039
  
@dawidwys What's your schedule to address the comments?
This is one of the real blockers for the 1.3.1 release, and I would like to 
put the first RC this week.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #4083: [FLINK-6742] Improve savepoint migration failure e...

[zentol]

GitHub user zentol opened a pull request:

https://github.com/apache/flink/pull/4083

[FLINK-6742] Improve savepoint migration failure error message

This PR improves the error messages if the savepoint migration fails 
because a stateful task was removed or the parallelism of stateful operator was 
changed.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/zentol/flink 6742

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/flink/pull/4083.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #4083


commit 6f701d17e7eb62a21f5c9466ba9acf8696ec9ab8
Author: zentol 
Date:   2017-06-07T10:03:21Z

[FLINK-6742] Improve savepoint migration failure error message

commit 38b07a7c4654a84b4370ed948be3ab76c28afad5
Author: zentol 
Date:   2017-06-07T10:03:57Z

[hotfix] Improve readability in SPV2#convertToOperatorStateSavepointV2




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Updated] (FLINK-6862) Tumble window rowtime not resolve at logic plan validation

[Mark You (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6862?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Mark You updated FLINK-6862:

Description: 
Following code sample work in version 1.2.1, but failed at 1.3.0
{code:title= TumblingWindow.java|borderStyle=solid}
public class TumblingWindow {

public static void main(String[] args) throws Exception {
List data = new ArrayList();
data.add(new Content(1L, "Hi"));
data.add(new Content(2L, "Hallo"));
data.add(new Content(3L, "Hello"));
data.add(new Content(4L, "Hello"));
data.add(new Content(7L, "Hello"));
data.add(new Content(8L, "Hello world"));
data.add(new Content(16L, "Hello world"));

final StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

final StreamTableEnvironment tableEnv = 
TableEnvironment.getTableEnvironment(env);

DataStream stream = env.fromCollection(data);

DataStream stream2 = stream.assignTimestampsAndWatermarks(
new 
BoundedOutOfOrdernessTimestampExtractor(Time.milliseconds(1)) {

/**
 * 
 */
private static final long serialVersionUID = 
410512296011057717L;

@Override
public long extractTimestamp(Content element) {
return element.getRecordTime();
}

});

Table table = tableEnv.fromDataStream(stream2);

table.window(Tumble.over("1.hours").on("rowtime").as("w")).groupBy("w").select("w.start,
 content.count");

env.execute();
}

public static class Content implements Serializable {

private long recordTime;
private String content;

public Content() {
super();
}

public Content(long recordTime, String content) {
super();
this.recordTime = recordTime;
this.content = content;
}

public long getRecordTime() {
return recordTime;
}

public void setRecordTime(long recordTime) {
this.recordTime = recordTime;
}

public String getContent() {
return content;
}

public void setContent(String content) {
this.content = content;
}

}

private class TimestampWithEqualWatermark implements 
AssignerWithPunctuatedWatermarks {

/**
 * 
 */
private static final long serialVersionUID = 1L;

@Override
public long extractTimestamp(Object[] element, long 
previousElementTimestamp) {
// TODO Auto-generated method stub
return (long) element[0];
}

@Override
public Watermark checkAndGetNextWatermark(Object[] lastElement, long 
extractedTimestamp) {
return new Watermark(extractedTimestamp);
}

}
}
{code}

Exception trace:
{noformat}
Exception in thread "main" org.apache.flink.table.api.ValidationException: 
Cannot resolve [rowtime] given input [content, recordTime].
at 
org.apache.flink.table.plan.logical.LogicalNode.failValidation(LogicalNode.scala:143)
at 
org.apache.flink.table.plan.logical.LogicalNode$$anonfun$validate$1.applyOrElse(LogicalNode.scala:86)
at 
org.apache.flink.table.plan.logical.LogicalNode$$anonfun$validate$1.applyOrElse(LogicalNode.scala:83)
at 
org.apache.flink.table.plan.TreeNode.postOrderTransform(TreeNode.scala:72)
at 
org.apache.flink.table.plan.logical.LogicalNode.org$apache$flink$table$plan$logical$LogicalNode$$expressionPostOrderTransform$1(LogicalNode.scala:119)
at 
org.apache.flink.table.plan.logical.LogicalNode$$anonfun$7$$anonfun$apply$1.apply(LogicalNode.scala:132)
at 
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
at 
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
at 
scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:245)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at 
org.apache.flink.table.plan.logical.LogicalNode$$anonfun$7.apply(LogicalNode.scala:131)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:370)
at scala.collection.Iterator$class.foreach(Iterator.scala:742)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1194)
at 
scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
at 
scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
at 

[jira] [Commented] (FLINK-6742) Improve error message when savepoint migration fails due to task removal

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6742:
---

GitHub user zentol opened a pull request:

https://github.com/apache/flink/pull/4083

[FLINK-6742] Improve savepoint migration failure error message

This PR improves the error messages if the savepoint migration fails 
because a stateful task was removed or the parallelism of stateful operator was 
changed.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/zentol/flink 6742

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/flink/pull/4083.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #4083


commit 6f701d17e7eb62a21f5c9466ba9acf8696ec9ab8
Author: zentol 
Date:   2017-06-07T10:03:21Z

[FLINK-6742] Improve savepoint migration failure error message

commit 38b07a7c4654a84b4370ed948be3ab76c28afad5
Author: zentol 
Date:   2017-06-07T10:03:57Z

[hotfix] Improve readability in SPV2#convertToOperatorStateSavepointV2




> Improve error message when savepoint migration fails due to task removal
> 
>
> Key: FLINK-6742
> URL: https://issues.apache.org/jira/browse/FLINK-6742
> Project: Flink
>  Issue Type: Bug
>  Components: State Backends, Checkpointing
>Affects Versions: 1.3.0
>Reporter: Gyula Fora
>Assignee: Chesnay Schepler
>Priority: Minor
>  Labels: flink-rel-1.3.1-blockers
>
> Caused by: java.lang.NullPointerException
>   at 
> org.apache.flink.runtime.checkpoint.savepoint.SavepointV2.convertToOperatorStateSavepointV2(SavepointV2.java:171)
>   at 
> org.apache.flink.runtime.checkpoint.savepoint.SavepointLoader.loadAndValidateSavepoint(SavepointLoader.java:75)
>   at 
> org.apache.flink.runtime.checkpoint.CheckpointCoordinator.restoreSavepoint(CheckpointCoordinator.java:1090)



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6788) Remove unused GenericFlatTypePostPass/AbstractSchema class

[Fabian Hueske (JIRA)]

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

Fabian Hueske commented on FLINK-6788:
--

I think it is fine to remove these classes. Their original intend was to be 
able to put different APIs on top of the optimizer. I think by now, it is quite 
unlikely that this will ever be done and if it would be done, they would 
probably need to be rewritten anyway.

> Remove unused GenericFlatTypePostPass/AbstractSchema class
> --
>
> Key: FLINK-6788
> URL: https://issues.apache.org/jira/browse/FLINK-6788
> Project: Flink
>  Issue Type: Improvement
>  Components: Optimizer
>Affects Versions: 1.4.0
>Reporter: Chesnay Schepler
>Priority: Trivial
>
> The {{AbstractSchema}} and {{GenericFlatTypePostPass}} classes in 
> {{org.apache.flink.optimizer.postpass}} are unused and could maybe be removed.
> [~fhueske] your thoughts?



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Assigned] (FLINK-6788) Remove unused GenericFlatTypePostPass/AbstractSchema class

[Chesnay Schepler (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6788?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Chesnay Schepler reassigned FLINK-6788:
---

Assignee: Miao Wang

> Remove unused GenericFlatTypePostPass/AbstractSchema class
> --
>
> Key: FLINK-6788
> URL: https://issues.apache.org/jira/browse/FLINK-6788
> Project: Flink
>  Issue Type: Improvement
>  Components: Optimizer
>Affects Versions: 1.4.0
>Reporter: Chesnay Schepler
>Assignee: Miao Wang
>Priority: Trivial
>
> The {{AbstractSchema}} and {{GenericFlatTypePostPass}} classes in 
> {{org.apache.flink.optimizer.postpass}} are unused and could maybe be removed.
> [~fhueske] your thoughts?



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6812) Elasticsearch 5 release artifacts not published to Maven central

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6812:
---

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/4048
  
I've tested the change with the release scripts, and the elasticsearch5 
connector was showing up.

I'll merge the change now.


> Elasticsearch 5 release artifacts not published to Maven central
> 
>
> Key: FLINK-6812
> URL: https://issues.apache.org/jira/browse/FLINK-6812
> Project: Flink
>  Issue Type: Bug
>  Components: ElasticSearch Connector
>Affects Versions: 1.3.0
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Robert Metzger
>Priority: Blocker
>  Labels: flink-rel-1.3.1-blockers
> Fix For: 1.3.1
>
>
> Release artifacts for the Elasticsearch 5 connector is not published to the 
> Maven Central. Elasticsearch 5 requires Java 8 at minimum, so for the release 
> we need to build with Java 8 for this.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Resolved] (FLINK-6812) Elasticsearch 5 release artifacts not published to Maven central

[Robert Metzger (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6812?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Robert Metzger resolved FLINK-6812.
---
   Resolution: Fixed
Fix Version/s: 1.4.0

Resolved for 1.4 in http://git-wip-us.apache.org/repos/asf/flink/commit/9ebd8c17
Resolved for 1.3 in http://git-wip-us.apache.org/repos/asf/flink/commit/97749585

> Elasticsearch 5 release artifacts not published to Maven central
> 
>
> Key: FLINK-6812
> URL: https://issues.apache.org/jira/browse/FLINK-6812
> Project: Flink
>  Issue Type: Bug
>  Components: ElasticSearch Connector
>Affects Versions: 1.3.0
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Robert Metzger
>Priority: Blocker
>  Labels: flink-rel-1.3.1-blockers
> Fix For: 1.3.1, 1.4.0
>
>
> Release artifacts for the Elasticsearch 5 connector is not published to the 
> Maven Central. Elasticsearch 5 requires Java 8 at minimum, so for the release 
> we need to build with Java 8 for this.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink issue #4081: [FLINK-6844] [scala] Implement compatibility methods for ...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/4081
  
Tested this with a stateful job that uses Scala collections as state.
Merging this ..


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #4079: [FLINK-6853] [DataStream] Let StreamRecordSerializer be c...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/4079
  
The test results in #4073 covers this change and results are green.
Since #4073 is not really a real blocker for 1.3.1, while this fix is to 
allow CEP migration from 1.1, I'll merge this independent of #4073 first. 


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6853) Migrating from Flink 1.1 fails for FlinkCEP

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6853:
---

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/4079
  
The test results in #4073 covers this change and results are green.
Since #4073 is not really a real blocker for 1.3.1, while this fix is to 
allow CEP migration from 1.1, I'll merge this independent of #4073 first. 


> Migrating from Flink 1.1 fails for FlinkCEP
> ---
>
> Key: FLINK-6853
> URL: https://issues.apache.org/jira/browse/FLINK-6853
> Project: Flink
>  Issue Type: Bug
>  Components: CEP
>Affects Versions: 1.3.0
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
>
> Migrating from Flink 1.1 fails for CEP, since in 1.1, the legacy 
> {{MultiplexingStreamRecordSerializer}} is used for stream  elements in the 
> serialized priority queue (via the {{PriorityQueueSerializer}}).
> In newer versions, the {{StreamElementSerializer}} is used instead. For this 
> to work, we need to implement the compatibility methods for 
> {{StreamElementSerializer}} such that it is also compatible with 
> configuration snapshots taken from the {{MultiplexingStreamRecordSerializer}}.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (FLINK-6742) Improve error message when savepoint migration fails due to task removal

[Chesnay Schepler (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6742?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Chesnay Schepler updated FLINK-6742:

Summary: Improve error message when savepoint migration fails due to task 
removal  (was: Savepoint conversion might fail if operators change)

> Improve error message when savepoint migration fails due to task removal
> 
>
> Key: FLINK-6742
> URL: https://issues.apache.org/jira/browse/FLINK-6742
> Project: Flink
>  Issue Type: Bug
>  Components: State Backends, Checkpointing
>Affects Versions: 1.3.0
>Reporter: Gyula Fora
>Priority: Minor
>  Labels: flink-rel-1.3.1-blockers
>
> Caused by: java.lang.NullPointerException
>   at 
> org.apache.flink.runtime.checkpoint.savepoint.SavepointV2.convertToOperatorStateSavepointV2(SavepointV2.java:171)
>   at 
> org.apache.flink.runtime.checkpoint.savepoint.SavepointLoader.loadAndValidateSavepoint(SavepointLoader.java:75)
>   at 
> org.apache.flink.runtime.checkpoint.CheckpointCoordinator.restoreSavepoint(CheckpointCoordinator.java:1090)



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Assigned] (FLINK-6742) Improve error message when savepoint migration fails due to task removal

[Chesnay Schepler (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-6742?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Chesnay Schepler reassigned FLINK-6742:
---

Assignee: Chesnay Schepler

> Improve error message when savepoint migration fails due to task removal
> 
>
> Key: FLINK-6742
> URL: https://issues.apache.org/jira/browse/FLINK-6742
> Project: Flink
>  Issue Type: Bug
>  Components: State Backends, Checkpointing
>Affects Versions: 1.3.0
>Reporter: Gyula Fora
>Assignee: Chesnay Schepler
>Priority: Minor
>  Labels: flink-rel-1.3.1-blockers
>
> Caused by: java.lang.NullPointerException
>   at 
> org.apache.flink.runtime.checkpoint.savepoint.SavepointV2.convertToOperatorStateSavepointV2(SavepointV2.java:171)
>   at 
> org.apache.flink.runtime.checkpoint.savepoint.SavepointLoader.loadAndValidateSavepoint(SavepointLoader.java:75)
>   at 
> org.apache.flink.runtime.checkpoint.CheckpointCoordinator.restoreSavepoint(CheckpointCoordinator.java:1090)



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6783) Wrongly extracted TypeInformations for WindowedStream::aggregate

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6783:
---

Github user dawidwys commented on the issue:

https://github.com/apache/flink/pull/4039
  
I will address them today. I am working on last comment with enabling the 
changes also for Partitioner.


> Wrongly extracted TypeInformations for WindowedStream::aggregate
> 
>
> Key: FLINK-6783
> URL: https://issues.apache.org/jira/browse/FLINK-6783
> Project: Flink
>  Issue Type: Bug
>  Components: Core, DataStream API
>Affects Versions: 1.3.0, 1.3.1
>Reporter: Dawid Wysakowicz
>Assignee: Dawid Wysakowicz
>Priority: Blocker
>  Labels: flink-rel-1.3.1-blockers
> Fix For: 1.3.1
>
>
> The following test fails because of wrongly acquired output type for 
> {{AggregateFunction}}:
> {code}
> @Test
> public void testAggregateWithWindowFunctionDifferentResultTypes() throws 
> Exception {
>   StreamExecutionEnvironment env = 
> StreamExecutionEnvironment.getExecutionEnvironment();
>   DataStream> source = 
> env.fromElements(Tuple2.of("hello", 1), Tuple2.of("hello", 2));
>   DataStream> window = source
>   .keyBy(new TupleKeySelector())
>   .window(TumblingEventTimeWindows.of(Time.of(1, 
> TimeUnit.SECONDS)))
>   .aggregate(new AggregateFunction, 
> Tuple2, String>() {
>   @Override
>   public Tuple2 createAccumulator() {
>   return Tuple2.of("", 0);
>   }
>   @Override
>   public void add(
>   Tuple2 value, Tuple2 Integer> accumulator) {
>   }
>   @Override
>   public String getResult(Tuple2 
> accumulator) {
>   return accumulator.f0;
>   }
>   @Override
>   public Tuple2 merge(
>   Tuple2 a, Tuple2 Integer> b) {
>   return Tuple2.of("", 0);
>   }
>   }, new WindowFunction, 
> String, TimeWindow>() {
>   @Override
>   public void apply(
>   String s,
>   TimeWindow window,
>   Iterable input,
>   Collector> out) 
> throws Exception {
>   out.collect(Tuple3.of("", "", 0));
>   }
>   });
>   OneInputTransformation, Tuple3 Integer>> transform =
>   (OneInputTransformation, Tuple3 String, Integer>>) window.getTransformation();
>   OneInputStreamOperator, Tuple3 Integer>> operator = transform.getOperator();
>   Assert.assertTrue(operator instanceof WindowOperator);
>   WindowOperator, ?, ?, ?> winOperator =
>   (WindowOperator, ?, ?, ?>) 
> operator;
>   Assert.assertTrue(winOperator.getTrigger() instanceof EventTimeTrigger);
>   Assert.assertTrue(winOperator.getWindowAssigner() instanceof 
> TumblingEventTimeWindows);
>   Assert.assertTrue(winOperator.getStateDescriptor() instanceof 
> AggregatingStateDescriptor);
>   processElementAndEnsureOutput(
>   operator, winOperator.getKeySelector(), 
> BasicTypeInfo.STRING_TYPE_INFO, new Tuple2<>("hello", 1));
> }
> {code}
> The test results in 
> {code}
> org.apache.flink.api.common.functions.InvalidTypesException: Input mismatch: 
> Tuple type expected.
>   at 
> org.apache.flink.api.java.typeutils.TypeExtractor.validateInputType(TypeExtractor.java:1157)
>   at 
> org.apache.flink.api.java.typeutils.TypeExtractor.getUnaryOperatorReturnType(TypeExtractor.java:451)
>   at 
> org.apache.flink.streaming.api.datastream.WindowedStream.aggregate(WindowedStream.java:855)
>   at 
> org.apache.flink.streaming.runtime.operators.windowing.WindowTranslationTest.testAggregateWithWindowFunctionDifferentResultTypes(WindowTranslationTest.java:702)
>   at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>   at 
> 

[jira] [Commented] (FLINK-6788) Remove unused GenericFlatTypePostPass/AbstractSchema class

[Chesnay Schepler (JIRA)]

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

Chesnay Schepler commented on FLINK-6788:
-

[~wm624] I've given you contributor permissions, you can now assign issues to 
yourself. I've also assigned this JIRA to you.

> Remove unused GenericFlatTypePostPass/AbstractSchema class
> --
>
> Key: FLINK-6788
> URL: https://issues.apache.org/jira/browse/FLINK-6788
> Project: Flink
>  Issue Type: Improvement
>  Components: Optimizer
>Affects Versions: 1.4.0
>Reporter: Chesnay Schepler
>Assignee: Miao Wang
>Priority: Trivial
>
> The {{AbstractSchema}} and {{GenericFlatTypePostPass}} classes in 
> {{org.apache.flink.optimizer.postpass}} are unused and could maybe be removed.
> [~fhueske] your thoughts?



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-6830) Add ITTests for savepoint migration from 1.3

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6830:
---

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/4059
  
Merging ...


> Add ITTests for savepoint migration from 1.3
> 
>
> Key: FLINK-6830
> URL: https://issues.apache.org/jira/browse/FLINK-6830
> Project: Flink
>  Issue Type: Test
>  Components: State Backends, Checkpointing
>Affects Versions: 1.3.0
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
> Fix For: 1.3.1
>
>
> Already with FLINK-6763 and FLINK-6764 we'll need to change the serialization 
> formats between 1.3.0 and 1.3.x.
> We probably should add the stateful job migration ITCases for restoring from 
> Flink 1.3.x now.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] flink issue #4059: [FLINK-6830] Add StatefulJobSavepointFrom13MigrationITCas...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/4059
  
Merging ...


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-6834) Over window doesn't support complex calculation

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on FLINK-6834:
---

Github user fhueske commented on a diff in the pull request:

https://github.com/apache/flink/pull/4070#discussion_r120586848
  
--- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/ProjectionTranslator.scala
 ---
@@ -226,30 +226,69 @@ object ProjectionTranslator {
   overWindows: Array[OverWindow],
   tEnv: TableEnvironment): Seq[Expression] = {
 
-def resolveOverWindow(unresolvedCall: UnresolvedOverCall): Expression 
= {
-
-  val overWindow = 
overWindows.find(_.alias.equals(unresolvedCall.alias))
-  if (overWindow.isDefined) {
-OverCall(
-  unresolvedCall.agg,
-  overWindow.get.partitionBy,
-  overWindow.get.orderBy,
-  overWindow.get.preceding,
-  overWindow.get.following)
-  } else {
-unresolvedCall
-  }
-}
+exprs.map(e => replaceOverCall(e, overWindows, tEnv))
+  }
 
-val projectList = new ListBuffer[Expression]
-exprs.foreach {
-  case Alias(u: UnresolvedOverCall, name, _) =>
-projectList += Alias(resolveOverWindow(u), name)
+  /**
+* Find and replace UnresolvedOverCall with OverCall
+*
+* @param exprthe expression to check
+* @return an expression with correct resolved OverCall
+*/
+  private def replaceOverCall(
+expr: Expression,
+overWindows: Array[OverWindow],
+tableEnv: TableEnvironment): Expression = {
+
+expr match {
   case u: UnresolvedOverCall =>
-projectList += resolveOverWindow(u)
-  case e: Expression => projectList += e
+val overWindow = overWindows.find(_.alias.equals(u.alias))
+if (overWindow.isDefined) {
+  OverCall(
+u.agg,
+overWindow.get.partitionBy,
+overWindow.get.orderBy,
+overWindow.get.preceding,
+overWindow.get.following)
+} else {
+  u
+}
+
+  case u: UnaryExpression =>
+val c = replaceOverCall(u.child, overWindows, tableEnv)
--- End diff --

Can you explain a bit more what you mean @sunjincheng121 ?


> Over window doesn't support complex calculation
> ---
>
> Key: FLINK-6834
> URL: https://issues.apache.org/jira/browse/FLINK-6834
> Project: Flink
>  Issue Type: Bug
>  Components: Table API & SQL
>Reporter: Jark Wu
>Assignee: Jark Wu
> Fix For: 1.4.0
>
>
> The following example
> {code}
> val windowedTable = table
>   .window(
> Over partitionBy 'c orderBy 'rowtime preceding UNBOUNDED_ROW as 'w)
>   .select('c, 'b, ('a.count over 'w) + 1)
> {code}
> will throw exception: 
> {code}
> org.apache.flink.table.api.ValidationException: Expression 
> UnresolvedOverCall(count('a),'w) failed on input check: Over window with 
> alias $alias could not be resolved.
>   at 
> org.apache.flink.table.plan.logical.LogicalNode.failValidation(LogicalNode.scala:143)
>   at 
> org.apache.flink.table.plan.logical.LogicalNode$$anonfun$validate$1.applyOrElse(LogicalNode.scala:89)
>   at 
> org.apache.flink.table.plan.logical.LogicalNode$$anonfun$validate$1.applyOrElse(LogicalNode.scala:83)
>   at 
> org.apache.flink.table.plan.TreeNode.postOrderTransform(TreeNode.scala:72)
>   at 
> org.apache.flink.table.plan.TreeNode$$anonfun$1.apply(TreeNode.scala:46)
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)