[jira] [Commented] (FLINK-4068) Move constant computations out of code-generated `flatMap` functions.

[ASF GitHub Bot (JIRA)]

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

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

Github user wuchong commented on the issue:

https://github.com/apache/flink/pull/2560
  
@twalthr  I add some Table API tests but skip some tests doesn't supported 
in Table API, such as `NULLIF`, `IN`, `EXTRACT`


> Move constant computations out of code-generated `flatMap` functions.
> -
>
> Key: FLINK-4068
> URL: https://issues.apache.org/jira/browse/FLINK-4068
> Project: Flink
>  Issue Type: Improvement
>  Components: Table API & SQL
>Affects Versions: 1.1.0
>Reporter: Fabian Hueske
>Assignee: Jark Wu
>
> The generated functions for expressions of the Table API or SQL include 
> constant computations.
> For instance the code generated for a predicate like:
> {code}
> myInt < (10 + 20)
> {code}
> looks roughly like:
> {code}
> public void flatMap(Row in, Collector out) {
>   Integer in1 = in.productElement(1);
>   int temp = 10 + 20;  
>   if (in1 < temp) {
> out.collect(in)
>   }
> }
> {code}
> In this example the computation of {{temp}} is constant and could be moved 
> out of the {{flatMap()}} method.
> The same might apply for generated function other than {{FlatMap}} as well.



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

[GitHub] flink issue #2560: [FLINK-4068] [table] Move constant computations out of co...

[wuchong]

Github user wuchong commented on the issue:

https://github.com/apache/flink/pull/2560
  
@twalthr  I add some Table API tests but skip some tests doesn't supported 
in Table API, such as `NULLIF`, `IN`, `EXTRACT`


---
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-4718) Confusing label in Parallel Streams Diagram

[ASF GitHub Bot (JIRA)]

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

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

GitHub user nderraugh opened a pull request:

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

'Id' is the key of the stream and not the id of the event.

FLINK-4718 Confusing label in Parallel Streams Diagram

Minor text change to clarify a mildly confusing label in a diagram.

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

$ git pull https://github.com/nderraugh/flink master

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

https://github.com/apache/flink/pull/2575.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 #2575


commit 479c8401f9680b4e0bf88af4c7b68033aeb54cdc
Author: Neil Derraugh 
Date:   2016-09-29T17:53:10Z

'Id' is the key of the stream and not the id of the event.




> Confusing label in Parallel Streams Diagram
> ---
>
> Key: FLINK-4718
> URL: https://issues.apache.org/jira/browse/FLINK-4718
> Project: Flink
>  Issue Type: Bug
>  Components: Project Website
>Affects Versions: 1.1.0
>Reporter: Neil Derraugh
>Priority: Trivial
>  Labels: newbie
> Fix For: 1.1.2
>
>
> The Event [id|timestamp] label in the Parallel Streams Diagram is confusing.  
> The 'id' is in fact the key of the stream and not the id of the event record. 
>  Hence we have B35 and B33.
> The 'id' label should be changed to key or key_id to better represent its 
> nature.



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

[GitHub] flink pull request #2575: 'Id' is the key of the stream and not the id of th...

[nderraugh]

GitHub user nderraugh opened a pull request:

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

'Id' is the key of the stream and not the id of the event.

FLINK-4718 Confusing label in Parallel Streams Diagram

Minor text change to clarify a mildly confusing label in a diagram.

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

$ git pull https://github.com/nderraugh/flink master

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

https://github.com/apache/flink/pull/2575.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 #2575


commit 479c8401f9680b4e0bf88af4c7b68033aeb54cdc
Author: Neil Derraugh 
Date:   2016-09-29T17:53:10Z

'Id' is the key of the stream and not the id of the event.




---
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-4068) Move constant computations out of code-generated `flatMap` functions.

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2560#discussion_r81270921
  
--- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkPlannerImpl.scala
 ---
@@ -60,6 +60,7 @@ class FlinkPlannerImpl(
   var root: RelRoot = _
 
   private def ready() {
+planner.setExecutor(config.getExecutor)
--- End diff --

Yes you are right ! This is the reason why Table API not work.


> Move constant computations out of code-generated `flatMap` functions.
> -
>
> Key: FLINK-4068
> URL: https://issues.apache.org/jira/browse/FLINK-4068
> Project: Flink
>  Issue Type: Improvement
>  Components: Table API & SQL
>Affects Versions: 1.1.0
>Reporter: Fabian Hueske
>Assignee: Jark Wu
>
> The generated functions for expressions of the Table API or SQL include 
> constant computations.
> For instance the code generated for a predicate like:
> {code}
> myInt < (10 + 20)
> {code}
> looks roughly like:
> {code}
> public void flatMap(Row in, Collector out) {
>   Integer in1 = in.productElement(1);
>   int temp = 10 + 20;  
>   if (in1 < temp) {
> out.collect(in)
>   }
> }
> {code}
> In this example the computation of {{temp}} is constant and could be moved 
> out of the {{flatMap()}} method.
> The same might apply for generated function other than {{FlatMap}} as well.



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

[GitHub] flink pull request #2560: [FLINK-4068] [table] Move constant computations ou...

[wuchong]

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

https://github.com/apache/flink/pull/2560#discussion_r81270921
  
--- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkPlannerImpl.scala
 ---
@@ -60,6 +60,7 @@ class FlinkPlannerImpl(
   var root: RelRoot = _
 
   private def ready() {
+planner.setExecutor(config.getExecutor)
--- End diff --

Yes you are right ! This is the reason why Table API not work.


---
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-4715) TaskManager should commit suicide after cancellation failure

[Zhijiang Wang (JIRA)]

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

Zhijiang Wang edited comment on FLINK-4715 at 9/30/16 2:31 AM:
---

Yes, we already experienced this problem in real production many times,  
because the user code can not be controlled. If the thread is waiting for 
synchronized lock or other cases, it can not be cancelled. We take the way that 
if the job master cancel the task failed many times, the job master will let 
the task manager exit itself.


was (Author: zjwang):
Yes, we already experienced this problem in real production many times,  
because the user code can not be controlled. If the thread is waiting for 
synchronized lock or other cases, it can not be cancelled, and the job master 
cancel the task failed many times, the job master will let the task manager 
exit itself.

> TaskManager should commit suicide after cancellation failure
> 
>
> Key: FLINK-4715
> URL: https://issues.apache.org/jira/browse/FLINK-4715
> Project: Flink
>  Issue Type: Improvement
>  Components: TaskManager
>Affects Versions: 1.2.0
>Reporter: Till Rohrmann
> Fix For: 1.2.0
>
>
> In case of a failed cancellation, e.g. the task cannot be cancelled after a 
> given time, the {{TaskManager}} should kill itself. That way we guarantee 
> that there is no resource leak. 
> This behaviour acts as a safety-net against faulty user code.



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

[jira] [Commented] (FLINK-4348) Implement slot allocation protocol with TaskExecutor

[ASF GitHub Bot (JIRA)]

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

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

Github user beyond1920 commented on the issue:

https://github.com/apache/flink/pull/2571
  
@mxm, What would happen under following cases: taskExecutor releases a 
registered slot,  then taskExecutor reports its latest slotReport to 
ResourceManager, ResourceManager should remove the old slot allocation from its 
own view and mark the slot free. So I think we should keep the following code 
in the old SlotManager `updateSlotStatus`:
_else {
// slot is reported empty

// check whether we also thought this slot is 
empty
if (allocationMap.isAllocated(slotId)) {
LOG.info("Slot allocation info 
mismatch! SlotID:{}, current:{}, reported:null",
slotId, 
allocationMap.getAllocationID(slotId));

// we thought the slot is in use, 
correct it
allocationMap.removeAllocation(slotId);

// we have a free slot!
handleFreeSlot(slot);
}
}_


> Implement slot allocation protocol with TaskExecutor
> 
>
> Key: FLINK-4348
> URL: https://issues.apache.org/jira/browse/FLINK-4348
> Project: Flink
>  Issue Type: Sub-task
>  Components: Cluster Management
>Reporter: Kurt Young
>Assignee: Maximilian Michels
>
> When slotManager finds a proper slot in the free pool for a slot request,  
> slotManager marks the slot as occupied, then tells the taskExecutor to give 
> the slot to the specified JobMaster. 
> when a slot request is sent to taskExecutor, it should contain following 
> parameters: AllocationID, JobID,  slotID, resourceManagerLeaderSessionID. 
> There exists 3 following possibilities of the response from taskExecutor, we 
> will discuss when each possibility happens and how to handle.
> 1. Ack request which means the taskExecutor gives the slot to the specified 
> jobMaster as expected.   
> 2. Decline request if the slot is already occupied by other AllocationID.  
> 3. Timeout which could caused by lost of request message or response message 
> or slow network transfer. 
> On the first occasion, ResourceManager need to do nothing. However, under the 
> second and third occasion, ResourceManager need to notify slotManager, 
> slotManager will verify and clear all the previous allocate information for 
> this slot request firstly, then try to find a proper slot for the slot 
> request again. This may cause some duplicate allocation, e.g. the slot 
> request to TaskManager is successful but the response is lost somehow, so we 
> may request a slot in another TaskManager, this causes two slots assigned to 
> one request, but it can be taken care of by rejecting registration at 
> JobMaster.
> There are still some question need to discuss in a step further.
> 1. Who send slotRequest to taskExecutor, SlotManager or ResourceManager? I 
> think it's better that SlotManager delegates the rpc call to ResourceManager 
> when SlotManager need to communicate with outside world.  ResourceManager 
> know which taskExecutor to send the request based on ResourceID. Besides this 
> RPC call which used to request slot to taskExecutor should not be a 
> RpcMethod,  because we hope only SlotManager has permission to call the 
> method, but the other component, for example JobMaster and TaskExecutor, 
> cannot call this method directly.
> 2. If JobMaster reject the slot offer from a TaskExecutor, the TaskExecutor 
> should notify the free slot to ResourceManager immediately, or wait for next 
> heartbeat sync. The advantage of first way is the resourceManager’s view 
> could be updated faster. The advantage of second way is save a RPC method in 
> ResourceManager.
> 3. There are two communication type. First, the slot request could be sent as 
> an ask operation where the response is returned as a future. Second, 
> resourceManager send the slot request in fire and forget way, the response 
> could be returned by an RPC call. I prefer the first one because it is more 
> simple and could save a RPC method in ResourceManager (for callback in the 
> second way).



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

[jira] [Created] (FLINK-4718) Confusing label in Parallel Streams Diagram

[Neil Derraugh (JIRA)]

Neil Derraugh created FLINK-4718:


 Summary: Confusing label in Parallel Streams Diagram
 Key: FLINK-4718
 URL: https://issues.apache.org/jira/browse/FLINK-4718
 Project: Flink
  Issue Type: Bug
  Components: Project Website
Affects Versions: 1.1.0
Reporter: Neil Derraugh
Priority: Trivial
 Fix For: 1.1.2


The Event [id|timestamp] label in the Parallel Streams Diagram is confusing.  
The 'id' is in fact the key of the stream and not the id of the event record.  
Hence we have B35 and B33.

The 'id' label should be changed to key or key_id to better represent its 
nature.



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

[GitHub] flink issue #2571: [FLINK-4348] Simplify logic of SlotManager

[beyond1920]

Github user beyond1920 commented on the issue:

https://github.com/apache/flink/pull/2571
  
@mxm, What would happen under following cases: taskExecutor releases a 
registered slot,  then taskExecutor reports its latest slotReport to 
ResourceManager, ResourceManager should remove the old slot allocation from its 
own view and mark the slot free. So I think we should keep the following code 
in the old SlotManager `updateSlotStatus`:
_else {
// slot is reported empty

// check whether we also thought this slot is 
empty
if (allocationMap.isAllocated(slotId)) {
LOG.info("Slot allocation info 
mismatch! SlotID:{}, current:{}, reported:null",
slotId, 
allocationMap.getAllocationID(slotId));

// we thought the slot is in use, 
correct it
allocationMap.removeAllocation(slotId);

// we have a free slot!
handleFreeSlot(slot);
}
}_


---
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-4715) TaskManager should commit suicide after cancellation failure

[Zhijiang Wang (JIRA)]

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

Zhijiang Wang commented on FLINK-4715:
--

Yes, we already experienced this problem in real production many times,  
because the user code can not be controlled. If the thread is waiting for 
synchronized lock or other cases, it can not be cancelled, and the job master 
cancel the task failed many times, the job master will let the task manager 
exit itself.

> TaskManager should commit suicide after cancellation failure
> 
>
> Key: FLINK-4715
> URL: https://issues.apache.org/jira/browse/FLINK-4715
> Project: Flink
>  Issue Type: Improvement
>  Components: TaskManager
>Affects Versions: 1.2.0
>Reporter: Till Rohrmann
> Fix For: 1.2.0
>
>
> In case of a failed cancellation, e.g. the task cannot be cancelled after a 
> given time, the {{TaskManager}} should kill itself. That way we guarantee 
> that there is no resource leak. 
> This behaviour acts as a safety-net against faulty user code.



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

[jira] [Commented] (FLINK-2254) Add Bipartite Graph Support for Gelly

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2564#discussion_r81246840
  
--- Diff: 
flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/BipartiteGraph.java
 ---
@@ -0,0 +1,231 @@
+/*
+ * 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.graph;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+/**
+ *
+ * Bipartite graph is a graph that contains two sets of vertices: top 
vertices and bottom vertices. Edges can only exist
+ * between a pair of vertices from different vertices sets. E.g. there can 
be no vertices between a pair
+ * of top vertices.
+ *
+ * Bipartite graph is useful to represent graphs with two sets of 
objects, like researchers and their publications,
+ * where an edge represents that a particular publication was authored by 
a particular author.
+ *
+ * Bipartite interface is different from {@link Graph} interface, so to 
apply algorithms that work on a regular graph
+ * a bipartite graph should be first converted into a {@link Graph} 
instance. This can be achieved by using
+ * {@link BipartiteGraph#topProjection(GroupReduceFunction)} or
+ * {@link BipartiteGraph#bottomProjection(GroupReduceFunction)} methods.
+ *
+ * @param  the key type of the top vertices
+ * @param  the key type of the bottom vertices
+ * @param  the top vertices value type
+ * @param  the bottom vertices value type
+ * @param  the edge value type
+ */
+public class BipartiteGraph {
+   private final ExecutionEnvironment context;
+   private final DataSet> topVertices;
+   private final DataSet> bottomVertices;
+   private final DataSet> edges;
+
+   private BipartiteGraph(
+   DataSet> topVertices,
+   DataSet> bottomVertices,
+   DataSet> edges,
+   ExecutionEnvironment context) {
+   this.topVertices = topVertices;
+   this.bottomVertices = bottomVertices;
+   this.edges = edges;
+   this.context = context;
+   }
+
+   /**
+* Create bipartite graph from datasets.
+*
+* @param topVertices  dataset of top vertices in the graph
+* @param bottomVertices dataset of bottom vertices in the graph
+* @param edges dataset of edges between vertices
+* @param context Flink execution context
+* @param  the key type of the top vertices
+* @param  the key type of the bottom vertices
+* @param  the top vertices value type
+* @param  the bottom vertices value type
+* @param  the edge value type
+* @return new bipartite graph created from provided datasets
+*/
+   public static  BipartiteGraph 
fromDataSet(
+   DataSet> topVertices,
+   DataSet> bottomVertices,
+   DataSet> edges,
+   ExecutionEnvironment context) {
+   return new BipartiteGraph<>(topVertices, bottomVertices, edges, 
context);
+   }
+
+   /**
+* Get dataset with top vertices.
+*
+* @return dataset with top vertices
+*/
+   public DataSet> getTopVertices() {

[GitHub] flink pull request #2564: [FLINK-2254] Add BipartiateGraph class

[mushketyk]

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

https://github.com/apache/flink/pull/2564#discussion_r81246840
  
--- Diff: 
flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/BipartiteGraph.java
 ---
@@ -0,0 +1,231 @@
+/*
+ * 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.graph;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+/**
+ *
+ * Bipartite graph is a graph that contains two sets of vertices: top 
vertices and bottom vertices. Edges can only exist
+ * between a pair of vertices from different vertices sets. E.g. there can 
be no vertices between a pair
+ * of top vertices.
+ *
+ * Bipartite graph is useful to represent graphs with two sets of 
objects, like researchers and their publications,
+ * where an edge represents that a particular publication was authored by 
a particular author.
+ *
+ * Bipartite interface is different from {@link Graph} interface, so to 
apply algorithms that work on a regular graph
+ * a bipartite graph should be first converted into a {@link Graph} 
instance. This can be achieved by using
+ * {@link BipartiteGraph#topProjection(GroupReduceFunction)} or
+ * {@link BipartiteGraph#bottomProjection(GroupReduceFunction)} methods.
+ *
+ * @param  the key type of the top vertices
+ * @param  the key type of the bottom vertices
+ * @param  the top vertices value type
+ * @param  the bottom vertices value type
+ * @param  the edge value type
+ */
+public class BipartiteGraph {
+   private final ExecutionEnvironment context;
+   private final DataSet> topVertices;
+   private final DataSet> bottomVertices;
+   private final DataSet> edges;
+
+   private BipartiteGraph(
+   DataSet> topVertices,
+   DataSet> bottomVertices,
+   DataSet> edges,
+   ExecutionEnvironment context) {
+   this.topVertices = topVertices;
+   this.bottomVertices = bottomVertices;
+   this.edges = edges;
+   this.context = context;
+   }
+
+   /**
+* Create bipartite graph from datasets.
+*
+* @param topVertices  dataset of top vertices in the graph
+* @param bottomVertices dataset of bottom vertices in the graph
+* @param edges dataset of edges between vertices
+* @param context Flink execution context
+* @param  the key type of the top vertices
+* @param  the key type of the bottom vertices
+* @param  the top vertices value type
+* @param  the bottom vertices value type
+* @param  the edge value type
+* @return new bipartite graph created from provided datasets
+*/
+   public static  BipartiteGraph 
fromDataSet(
+   DataSet> topVertices,
+   DataSet> bottomVertices,
+   DataSet> edges,
+   ExecutionEnvironment context) {
+   return new BipartiteGraph<>(topVertices, bottomVertices, edges, 
context);
+   }
+
+   /**
+* Get dataset with top vertices.
+*
+* @return dataset with top vertices
+*/
+   public DataSet> getTopVertices() {
+   return topVertices;
+   }
+
+   /**
+* Get dataset with bottom vertices.
+*
+* @return dataset with bottom vertices
+*/
+   public DataSet> getBottomVertices() {
  

[jira] [Commented] (FLINK-2254) Add Bipartite Graph Support for Gelly

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2564#discussion_r81246465
  
--- Diff: 
flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/Edge.java ---
@@ -34,10 +34,10 @@
 
public Edge(){}
 
-   public Edge(K src, K trg, V val) {
-   this.f0 = src;
-   this.f1 = trg;
-   this.f2 = val;
+   public Edge(K source, K target, V value) {
--- End diff --

To make them consistent with naming style in other classes.
Do you suggest to revert this?


> Add Bipartite Graph Support for Gelly
> -
>
> Key: FLINK-2254
> URL: https://issues.apache.org/jira/browse/FLINK-2254
> Project: Flink
>  Issue Type: New Feature
>  Components: Gelly
>Affects Versions: 0.10.0
>Reporter: Andra Lungu
>Assignee: Ivan Mushketyk
>  Labels: requires-design-doc
>
> A bipartite graph is a graph for which the set of vertices can be divided 
> into two disjoint sets such that each edge having a source vertex in the 
> first set, will have a target vertex in the second set. We would like to 
> support efficient operations for this type of graphs along with a set of 
> metrics(http://jponnela.com/web_documents/twomode.pdf). 



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

[GitHub] flink pull request #2564: [FLINK-2254] Add BipartiateGraph class

[mushketyk]

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

https://github.com/apache/flink/pull/2564#discussion_r81246465
  
--- Diff: 
flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/Edge.java ---
@@ -34,10 +34,10 @@
 
public Edge(){}
 
-   public Edge(K src, K trg, V val) {
-   this.f0 = src;
-   this.f1 = trg;
-   this.f2 = val;
+   public Edge(K source, K target, V value) {
--- End diff --

To make them consistent with naming style in other classes.
Do you suggest to revert 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.
---

[jira] [Commented] (FLINK-2254) Add Bipartite Graph Support for Gelly

[ASF GitHub Bot (JIRA)]

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

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

Github user mushketyk commented on the issue:

https://github.com/apache/flink/pull/2564
  
Hi @greghogan,

I like your ideas about providing different API for projections. This 
should be better than my approach. 
@vasia What do you think about this?


> Add Bipartite Graph Support for Gelly
> -
>
> Key: FLINK-2254
> URL: https://issues.apache.org/jira/browse/FLINK-2254
> Project: Flink
>  Issue Type: New Feature
>  Components: Gelly
>Affects Versions: 0.10.0
>Reporter: Andra Lungu
>Assignee: Ivan Mushketyk
>  Labels: requires-design-doc
>
> A bipartite graph is a graph for which the set of vertices can be divided 
> into two disjoint sets such that each edge having a source vertex in the 
> first set, will have a target vertex in the second set. We would like to 
> support efficient operations for this type of graphs along with a set of 
> metrics(http://jponnela.com/web_documents/twomode.pdf). 



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

[GitHub] flink issue #2564: [FLINK-2254] Add BipartiateGraph class

[mushketyk]

Github user mushketyk commented on the issue:

https://github.com/apache/flink/pull/2564
  
Hi @greghogan,

I like your ideas about providing different API for projections. This 
should be better than my approach. 
@vasia What do you think about 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.
---

[jira] [Commented] (FLINK-2254) Add Bipartite Graph Support for Gelly

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2564#discussion_r81245767
  
--- Diff: 
flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
 ---
@@ -480,7 +480,8 @@ protected static File asFile(String path) {
}
}

-   assertEquals("Wrong number of elements result", 
expectedStrings.length, resultStrings.length);
+   assertEquals(String.format("Wrong number of elements result. 
Expected: %s. Result: %s.", Arrays.toString(expectedStrings), 
Arrays.toString(resultStrings)),
--- End diff --

The issue here is that it compares lengths of objects and therefore JUnit 
only prints compared numbers (say 2 and 0) and not content of arrays.


> Add Bipartite Graph Support for Gelly
> -
>
> Key: FLINK-2254
> URL: https://issues.apache.org/jira/browse/FLINK-2254
> Project: Flink
>  Issue Type: New Feature
>  Components: Gelly
>Affects Versions: 0.10.0
>Reporter: Andra Lungu
>Assignee: Ivan Mushketyk
>  Labels: requires-design-doc
>
> A bipartite graph is a graph for which the set of vertices can be divided 
> into two disjoint sets such that each edge having a source vertex in the 
> first set, will have a target vertex in the second set. We would like to 
> support efficient operations for this type of graphs along with a set of 
> metrics(http://jponnela.com/web_documents/twomode.pdf). 



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

[GitHub] flink pull request #2564: [FLINK-2254] Add BipartiateGraph class

[mushketyk]

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

https://github.com/apache/flink/pull/2564#discussion_r81245767
  
--- Diff: 
flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
 ---
@@ -480,7 +480,8 @@ protected static File asFile(String path) {
}
}

-   assertEquals("Wrong number of elements result", 
expectedStrings.length, resultStrings.length);
+   assertEquals(String.format("Wrong number of elements result. 
Expected: %s. Result: %s.", Arrays.toString(expectedStrings), 
Arrays.toString(resultStrings)),
--- End diff --

The issue here is that it compares lengths of objects and therefore JUnit 
only prints compared numbers (say 2 and 0) and not content of arrays.


---
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-4329) Fix Streaming File Source Timestamps/Watermarks Handling

[ASF GitHub Bot (JIRA)]

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

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

Github user kl0u commented on the issue:

https://github.com/apache/flink/pull/2546
  
Hi @StephanEwen . If I understand correctly, your suggestion is to make the 
test something like the following: 1) put the split in the reader 2) read the 
split 3) when the split finishes update the time in the provider 4) observe the 
time in the output elements. If this is the case, then the problem is that the 
reader just puts the split in a queue, and this is picked up by another thread 
that reads it. In this context, there is no way of knowing when the reading 
thread has finished reading the split and goes to the next one. So step 3) 
cannot be synchronized correctly. This is the reason I am just having a thread 
in the test that tries (without guarantees - the race condition you mentioned) 
to update the time while the reader is still reading. Any suggestions are 
welcome.


> Fix Streaming File Source Timestamps/Watermarks Handling
> 
>
> Key: FLINK-4329
> URL: https://issues.apache.org/jira/browse/FLINK-4329
> Project: Flink
>  Issue Type: Bug
>  Components: Streaming Connectors
>Affects Versions: 1.1.0
>Reporter: Aljoscha Krettek
>Assignee: Kostas Kloudas
> Fix For: 1.2.0, 1.1.3
>
>
> The {{ContinuousFileReaderOperator}} does not correctly deal with watermarks, 
> i.e. they are just passed through. This means that when the 
> {{ContinuousFileMonitoringFunction}} closes and emits a {{Long.MAX_VALUE}} 
> that watermark can "overtake" the records that are to be emitted in the 
> {{ContinuousFileReaderOperator}}. Together with the new "allowed lateness" 
> setting in window operator this can lead to elements being dropped as late.
> Also, {{ContinuousFileReaderOperator}} does not correctly assign ingestion 
> timestamps since it is not technically a source but looks like one to the 
> user.



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

[GitHub] flink issue #2546: [FLINK-4329] Fix Streaming File Source Timestamps/Waterma...

[kl0u]

Github user kl0u commented on the issue:

https://github.com/apache/flink/pull/2546
  
Hi @StephanEwen . If I understand correctly, your suggestion is to make the 
test something like the following: 1) put the split in the reader 2) read the 
split 3) when the split finishes update the time in the provider 4) observe the 
time in the output elements. If this is the case, then the problem is that the 
reader just puts the split in a queue, and this is picked up by another thread 
that reads it. In this context, there is no way of knowing when the reading 
thread has finished reading the split and goes to the next one. So step 3) 
cannot be synchronized correctly. This is the reason I am just having a thread 
in the test that tries (without guarantees - the race condition you mentioned) 
to update the time while the reader is still reading. Any suggestions are 
welcome.


---
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-2254) Add Bipartite Graph Support for Gelly

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2564#discussion_r81225166
  
--- Diff: 
flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
 ---
@@ -480,7 +480,8 @@ protected static File asFile(String path) {
}
}

-   assertEquals("Wrong number of elements result", 
expectedStrings.length, resultStrings.length);
+   assertEquals(String.format("Wrong number of elements result. 
Expected: %s. Result: %s.", Arrays.toString(expectedStrings), 
Arrays.toString(resultStrings)),
--- End diff --

Doesn't IntelliJ offer to view the different results?


> Add Bipartite Graph Support for Gelly
> -
>
> Key: FLINK-2254
> URL: https://issues.apache.org/jira/browse/FLINK-2254
> Project: Flink
>  Issue Type: New Feature
>  Components: Gelly
>Affects Versions: 0.10.0
>Reporter: Andra Lungu
>Assignee: Ivan Mushketyk
>  Labels: requires-design-doc
>
> A bipartite graph is a graph for which the set of vertices can be divided 
> into two disjoint sets such that each edge having a source vertex in the 
> first set, will have a target vertex in the second set. We would like to 
> support efficient operations for this type of graphs along with a set of 
> metrics(http://jponnela.com/web_documents/twomode.pdf). 



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

[jira] [Commented] (FLINK-2254) Add Bipartite Graph Support for Gelly

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2564#discussion_r81226805
  
--- Diff: 
flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/BipartiteGraph.java
 ---
@@ -0,0 +1,231 @@
+/*
+ * 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.graph;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+/**
+ *
+ * Bipartite graph is a graph that contains two sets of vertices: top 
vertices and bottom vertices. Edges can only exist
+ * between a pair of vertices from different vertices sets. E.g. there can 
be no vertices between a pair
+ * of top vertices.
+ *
+ * Bipartite graph is useful to represent graphs with two sets of 
objects, like researchers and their publications,
+ * where an edge represents that a particular publication was authored by 
a particular author.
+ *
+ * Bipartite interface is different from {@link Graph} interface, so to 
apply algorithms that work on a regular graph
+ * a bipartite graph should be first converted into a {@link Graph} 
instance. This can be achieved by using
+ * {@link BipartiteGraph#topProjection(GroupReduceFunction)} or
+ * {@link BipartiteGraph#bottomProjection(GroupReduceFunction)} methods.
+ *
+ * @param  the key type of the top vertices
+ * @param  the key type of the bottom vertices
+ * @param  the top vertices value type
+ * @param  the bottom vertices value type
+ * @param  the edge value type
+ */
+public class BipartiteGraph {
+   private final ExecutionEnvironment context;
+   private final DataSet> topVertices;
+   private final DataSet> bottomVertices;
+   private final DataSet> edges;
+
+   private BipartiteGraph(
+   DataSet> topVertices,
+   DataSet> bottomVertices,
+   DataSet> edges,
+   ExecutionEnvironment context) {
+   this.topVertices = topVertices;
+   this.bottomVertices = bottomVertices;
+   this.edges = edges;
+   this.context = context;
+   }
+
+   /**
+* Create bipartite graph from datasets.
+*
+* @param topVertices  dataset of top vertices in the graph
+* @param bottomVertices dataset of bottom vertices in the graph
+* @param edges dataset of edges between vertices
+* @param context Flink execution context
+* @param  the key type of the top vertices
+* @param  the key type of the bottom vertices
+* @param  the top vertices value type
+* @param  the bottom vertices value type
+* @param  the edge value type
+* @return new bipartite graph created from provided datasets
+*/
+   public static  BipartiteGraph 
fromDataSet(
+   DataSet> topVertices,
+   DataSet> bottomVertices,
+   DataSet> edges,
+   ExecutionEnvironment context) {
+   return new BipartiteGraph<>(topVertices, bottomVertices, edges, 
context);
+   }
+
+   /**
+* Get dataset with top vertices.
+*
+* @return dataset with top vertices
+*/
+   public DataSet> getTopVertices() {

[GitHub] flink pull request #2564: [FLINK-2254] Add BipartiateGraph class

[greghogan]

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

https://github.com/apache/flink/pull/2564#discussion_r81225166
  
--- Diff: 
flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
 ---
@@ -480,7 +480,8 @@ protected static File asFile(String path) {
}
}

-   assertEquals("Wrong number of elements result", 
expectedStrings.length, resultStrings.length);
+   assertEquals(String.format("Wrong number of elements result. 
Expected: %s. Result: %s.", Arrays.toString(expectedStrings), 
Arrays.toString(resultStrings)),
--- End diff --

Doesn't IntelliJ offer to view the different results?


---
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 #2564: [FLINK-2254] Add BipartiateGraph class

[greghogan]

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

https://github.com/apache/flink/pull/2564#discussion_r81226805
  
--- Diff: 
flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/BipartiteGraph.java
 ---
@@ -0,0 +1,231 @@
+/*
+ * 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.graph;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+/**
+ *
+ * Bipartite graph is a graph that contains two sets of vertices: top 
vertices and bottom vertices. Edges can only exist
+ * between a pair of vertices from different vertices sets. E.g. there can 
be no vertices between a pair
+ * of top vertices.
+ *
+ * Bipartite graph is useful to represent graphs with two sets of 
objects, like researchers and their publications,
+ * where an edge represents that a particular publication was authored by 
a particular author.
+ *
+ * Bipartite interface is different from {@link Graph} interface, so to 
apply algorithms that work on a regular graph
+ * a bipartite graph should be first converted into a {@link Graph} 
instance. This can be achieved by using
+ * {@link BipartiteGraph#topProjection(GroupReduceFunction)} or
+ * {@link BipartiteGraph#bottomProjection(GroupReduceFunction)} methods.
+ *
+ * @param  the key type of the top vertices
+ * @param  the key type of the bottom vertices
+ * @param  the top vertices value type
+ * @param  the bottom vertices value type
+ * @param  the edge value type
+ */
+public class BipartiteGraph {
+   private final ExecutionEnvironment context;
+   private final DataSet> topVertices;
+   private final DataSet> bottomVertices;
+   private final DataSet> edges;
+
+   private BipartiteGraph(
+   DataSet> topVertices,
+   DataSet> bottomVertices,
+   DataSet> edges,
+   ExecutionEnvironment context) {
+   this.topVertices = topVertices;
+   this.bottomVertices = bottomVertices;
+   this.edges = edges;
+   this.context = context;
+   }
+
+   /**
+* Create bipartite graph from datasets.
+*
+* @param topVertices  dataset of top vertices in the graph
+* @param bottomVertices dataset of bottom vertices in the graph
+* @param edges dataset of edges between vertices
+* @param context Flink execution context
+* @param  the key type of the top vertices
+* @param  the key type of the bottom vertices
+* @param  the top vertices value type
+* @param  the bottom vertices value type
+* @param  the edge value type
+* @return new bipartite graph created from provided datasets
+*/
+   public static  BipartiteGraph 
fromDataSet(
+   DataSet> topVertices,
+   DataSet> bottomVertices,
+   DataSet> edges,
+   ExecutionEnvironment context) {
+   return new BipartiteGraph<>(topVertices, bottomVertices, edges, 
context);
+   }
+
+   /**
+* Get dataset with top vertices.
+*
+* @return dataset with top vertices
+*/
+   public DataSet> getTopVertices() {
+   return topVertices;
+   }
+
+   /**
+* Get dataset with bottom vertices.
+*
+* @return dataset with bottom vertices
+*/
+   public DataSet> getBottomVertices() {
  

[GitHub] flink pull request #2564: [FLINK-2254] Add BipartiateGraph class

[vasia]

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

https://github.com/apache/flink/pull/2564#discussion_r81217922
  
--- Diff: 
flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/BipartiteGraph.java
 ---
@@ -0,0 +1,231 @@
+/*
+ * 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.graph;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+/**
+ *
+ * Bipartite graph is a graph that contains two sets of vertices: top 
vertices and bottom vertices. Edges can only exist
--- End diff --

I would rephrase that to "... a graph whose vertices can be divided into 
two disjoint sets"


---
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 #2564: [FLINK-2254] Add BipartiateGraph class

[vasia]

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

https://github.com/apache/flink/pull/2564#discussion_r81218538
  
--- Diff: 
flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/Edge.java ---
@@ -34,10 +34,10 @@
 
public Edge(){}
 
-   public Edge(K src, K trg, V val) {
-   this.f0 = src;
-   this.f1 = trg;
-   this.f2 = val;
+   public Edge(K source, K target, V value) {
--- End diff --

Why did you change these?


---
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 #2564: [FLINK-2254] Add BipartiateGraph class

[vasia]

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

https://github.com/apache/flink/pull/2564#discussion_r81218403
  
--- Diff: 
flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/BipartiteGraph.java
 ---
@@ -0,0 +1,231 @@
+/*
+ * 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.graph;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+/**
+ *
+ * Bipartite graph is a graph that contains two sets of vertices: top 
vertices and bottom vertices. Edges can only exist
+ * between a pair of vertices from different vertices sets. E.g. there can 
be no vertices between a pair
+ * of top vertices.
+ *
+ * Bipartite graph is useful to represent graphs with two sets of 
objects, like researchers and their publications,
+ * where an edge represents that a particular publication was authored by 
a particular author.
+ *
+ * Bipartite interface is different from {@link Graph} interface, so to 
apply algorithms that work on a regular graph
+ * a bipartite graph should be first converted into a {@link Graph} 
instance. This can be achieved by using
+ * {@link BipartiteGraph#topProjection(GroupReduceFunction)} or
+ * {@link BipartiteGraph#bottomProjection(GroupReduceFunction)} methods.
+ *
+ * @param  the key type of the top vertices
+ * @param  the key type of the bottom vertices
+ * @param  the top vertices value type
+ * @param  the bottom vertices value type
+ * @param  the edge value type
+ */
+public class BipartiteGraph {
+   private final ExecutionEnvironment context;
+   private final DataSet> topVertices;
+   private final DataSet> bottomVertices;
+   private final DataSet> edges;
+
+   private BipartiteGraph(
+   DataSet> topVertices,
+   DataSet> bottomVertices,
+   DataSet> edges,
+   ExecutionEnvironment context) {
+   this.topVertices = topVertices;
+   this.bottomVertices = bottomVertices;
+   this.edges = edges;
+   this.context = context;
+   }
+
+   /**
+* Create bipartite graph from datasets.
+*
+* @param topVertices  dataset of top vertices in the graph
+* @param bottomVertices dataset of bottom vertices in the graph
+* @param edges dataset of edges between vertices
+* @param context Flink execution context
+* @param  the key type of the top vertices
+* @param  the key type of the bottom vertices
+* @param  the top vertices value type
+* @param  the bottom vertices value type
+* @param  the edge value type
+* @return new bipartite graph created from provided datasets
+*/
+   public static  BipartiteGraph 
fromDataSet(
+   DataSet> topVertices,
+   DataSet> bottomVertices,
+   DataSet> edges,
+   ExecutionEnvironment context) {
+   return new BipartiteGraph<>(topVertices, bottomVertices, edges, 
context);
+   }
+
+   /**
+* Get dataset with top vertices.
+*
+* @return dataset with top vertices
+*/
+   public DataSet> getTopVertices() {
+   return topVertices;
+   }
+
+   /**
+* Get dataset with bottom vertices.
+*
+* @return dataset with bottom vertices
+*/
+   public DataSet> getBottomVertices() {
+ 

[GitHub] flink pull request #2564: [FLINK-2254] Add BipartiateGraph class

[vasia]

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

https://github.com/apache/flink/pull/2564#discussion_r81218056
  
--- Diff: 
flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/BipartiteGraph.java
 ---
@@ -0,0 +1,231 @@
+/*
+ * 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.graph;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+/**
+ *
+ * Bipartite graph is a graph that contains two sets of vertices: top 
vertices and bottom vertices. Edges can only exist
+ * between a pair of vertices from different vertices sets. E.g. there can 
be no vertices between a pair
+ * of top vertices.
+ *
+ * Bipartite graph is useful to represent graphs with two sets of 
objects, like researchers and their publications,
--- End diff --

Bipartite graphs are useful...


---
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 #2564: [FLINK-2254] Add BipartiateGraph class

[vasia]

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

https://github.com/apache/flink/pull/2564#discussion_r81218490
  
--- Diff: 
flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/BipartiteGraph.java
 ---
@@ -0,0 +1,231 @@
+/*
+ * 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.graph;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+/**
+ *
+ * Bipartite graph is a graph that contains two sets of vertices: top 
vertices and bottom vertices. Edges can only exist
+ * between a pair of vertices from different vertices sets. E.g. there can 
be no vertices between a pair
+ * of top vertices.
+ *
+ * Bipartite graph is useful to represent graphs with two sets of 
objects, like researchers and their publications,
+ * where an edge represents that a particular publication was authored by 
a particular author.
+ *
+ * Bipartite interface is different from {@link Graph} interface, so to 
apply algorithms that work on a regular graph
+ * a bipartite graph should be first converted into a {@link Graph} 
instance. This can be achieved by using
+ * {@link BipartiteGraph#topProjection(GroupReduceFunction)} or
+ * {@link BipartiteGraph#bottomProjection(GroupReduceFunction)} methods.
+ *
+ * @param  the key type of the top vertices
+ * @param  the key type of the bottom vertices
+ * @param  the top vertices value type
+ * @param  the bottom vertices value type
+ * @param  the edge value type
+ */
+public class BipartiteGraph {
+   private final ExecutionEnvironment context;
+   private final DataSet> topVertices;
+   private final DataSet> bottomVertices;
+   private final DataSet> edges;
+
+   private BipartiteGraph(
+   DataSet> topVertices,
+   DataSet> bottomVertices,
+   DataSet> edges,
+   ExecutionEnvironment context) {
+   this.topVertices = topVertices;
+   this.bottomVertices = bottomVertices;
+   this.edges = edges;
+   this.context = context;
+   }
+
+   /**
+* Create bipartite graph from datasets.
+*
+* @param topVertices  dataset of top vertices in the graph
+* @param bottomVertices dataset of bottom vertices in the graph
+* @param edges dataset of edges between vertices
+* @param context Flink execution context
+* @param  the key type of the top vertices
+* @param  the key type of the bottom vertices
+* @param  the top vertices value type
+* @param  the bottom vertices value type
+* @param  the edge value type
+* @return new bipartite graph created from provided datasets
+*/
+   public static  BipartiteGraph 
fromDataSet(
+   DataSet> topVertices,
+   DataSet> bottomVertices,
+   DataSet> edges,
+   ExecutionEnvironment context) {
+   return new BipartiteGraph<>(topVertices, bottomVertices, edges, 
context);
+   }
+
+   /**
+* Get dataset with top vertices.
+*
+* @return dataset with top vertices
+*/
+   public DataSet> getTopVertices() {
+   return topVertices;
+   }
+
+   /**
+* Get dataset with bottom vertices.
+*
+* @return dataset with bottom vertices
+*/
+   public DataSet> getBottomVertices() {
+ 

[jira] [Commented] (FLINK-2254) Add Bipartite Graph Support for Gelly

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2564#discussion_r81218403
  
--- Diff: 
flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/BipartiteGraph.java
 ---
@@ -0,0 +1,231 @@
+/*
+ * 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.graph;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+/**
+ *
+ * Bipartite graph is a graph that contains two sets of vertices: top 
vertices and bottom vertices. Edges can only exist
+ * between a pair of vertices from different vertices sets. E.g. there can 
be no vertices between a pair
+ * of top vertices.
+ *
+ * Bipartite graph is useful to represent graphs with two sets of 
objects, like researchers and their publications,
+ * where an edge represents that a particular publication was authored by 
a particular author.
+ *
+ * Bipartite interface is different from {@link Graph} interface, so to 
apply algorithms that work on a regular graph
+ * a bipartite graph should be first converted into a {@link Graph} 
instance. This can be achieved by using
+ * {@link BipartiteGraph#topProjection(GroupReduceFunction)} or
+ * {@link BipartiteGraph#bottomProjection(GroupReduceFunction)} methods.
+ *
+ * @param  the key type of the top vertices
+ * @param  the key type of the bottom vertices
+ * @param  the top vertices value type
+ * @param  the bottom vertices value type
+ * @param  the edge value type
+ */
+public class BipartiteGraph {
+   private final ExecutionEnvironment context;
+   private final DataSet> topVertices;
+   private final DataSet> bottomVertices;
+   private final DataSet> edges;
+
+   private BipartiteGraph(
+   DataSet> topVertices,
+   DataSet> bottomVertices,
+   DataSet> edges,
+   ExecutionEnvironment context) {
+   this.topVertices = topVertices;
+   this.bottomVertices = bottomVertices;
+   this.edges = edges;
+   this.context = context;
+   }
+
+   /**
+* Create bipartite graph from datasets.
+*
+* @param topVertices  dataset of top vertices in the graph
+* @param bottomVertices dataset of bottom vertices in the graph
+* @param edges dataset of edges between vertices
+* @param context Flink execution context
+* @param  the key type of the top vertices
+* @param  the key type of the bottom vertices
+* @param  the top vertices value type
+* @param  the bottom vertices value type
+* @param  the edge value type
+* @return new bipartite graph created from provided datasets
+*/
+   public static  BipartiteGraph 
fromDataSet(
+   DataSet> topVertices,
+   DataSet> bottomVertices,
+   DataSet> edges,
+   ExecutionEnvironment context) {
+   return new BipartiteGraph<>(topVertices, bottomVertices, edges, 
context);
+   }
+
+   /**
+* Get dataset with top vertices.
+*
+* @return dataset with top vertices
+*/
+   public DataSet> getTopVertices() {
+   

[jira] [Commented] (FLINK-2254) Add Bipartite Graph Support for Gelly

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2564#discussion_r81217922
  
--- Diff: 
flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/BipartiteGraph.java
 ---
@@ -0,0 +1,231 @@
+/*
+ * 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.graph;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+/**
+ *
+ * Bipartite graph is a graph that contains two sets of vertices: top 
vertices and bottom vertices. Edges can only exist
--- End diff --

I would rephrase that to "... a graph whose vertices can be divided into 
two disjoint sets"


> Add Bipartite Graph Support for Gelly
> -
>
> Key: FLINK-2254
> URL: https://issues.apache.org/jira/browse/FLINK-2254
> Project: Flink
>  Issue Type: New Feature
>  Components: Gelly
>Affects Versions: 0.10.0
>Reporter: Andra Lungu
>Assignee: Ivan Mushketyk
>  Labels: requires-design-doc
>
> A bipartite graph is a graph for which the set of vertices can be divided 
> into two disjoint sets such that each edge having a source vertex in the 
> first set, will have a target vertex in the second set. We would like to 
> support efficient operations for this type of graphs along with a set of 
> metrics(http://jponnela.com/web_documents/twomode.pdf). 



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

[jira] [Commented] (FLINK-2254) Add Bipartite Graph Support for Gelly

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2564#discussion_r81218538
  
--- Diff: 
flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/Edge.java ---
@@ -34,10 +34,10 @@
 
public Edge(){}
 
-   public Edge(K src, K trg, V val) {
-   this.f0 = src;
-   this.f1 = trg;
-   this.f2 = val;
+   public Edge(K source, K target, V value) {
--- End diff --

Why did you change these?


> Add Bipartite Graph Support for Gelly
> -
>
> Key: FLINK-2254
> URL: https://issues.apache.org/jira/browse/FLINK-2254
> Project: Flink
>  Issue Type: New Feature
>  Components: Gelly
>Affects Versions: 0.10.0
>Reporter: Andra Lungu
>Assignee: Ivan Mushketyk
>  Labels: requires-design-doc
>
> A bipartite graph is a graph for which the set of vertices can be divided 
> into two disjoint sets such that each edge having a source vertex in the 
> first set, will have a target vertex in the second set. We would like to 
> support efficient operations for this type of graphs along with a set of 
> metrics(http://jponnela.com/web_documents/twomode.pdf). 



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

[jira] [Commented] (FLINK-2254) Add Bipartite Graph Support for Gelly

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2564#discussion_r81218056
  
--- Diff: 
flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/BipartiteGraph.java
 ---
@@ -0,0 +1,231 @@
+/*
+ * 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.graph;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+/**
+ *
+ * Bipartite graph is a graph that contains two sets of vertices: top 
vertices and bottom vertices. Edges can only exist
+ * between a pair of vertices from different vertices sets. E.g. there can 
be no vertices between a pair
+ * of top vertices.
+ *
+ * Bipartite graph is useful to represent graphs with two sets of 
objects, like researchers and their publications,
--- End diff --

Bipartite graphs are useful...


> Add Bipartite Graph Support for Gelly
> -
>
> Key: FLINK-2254
> URL: https://issues.apache.org/jira/browse/FLINK-2254
> Project: Flink
>  Issue Type: New Feature
>  Components: Gelly
>Affects Versions: 0.10.0
>Reporter: Andra Lungu
>Assignee: Ivan Mushketyk
>  Labels: requires-design-doc
>
> A bipartite graph is a graph for which the set of vertices can be divided 
> into two disjoint sets such that each edge having a source vertex in the 
> first set, will have a target vertex in the second set. We would like to 
> support efficient operations for this type of graphs along with a set of 
> metrics(http://jponnela.com/web_documents/twomode.pdf). 



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

[jira] [Commented] (FLINK-2254) Add Bipartite Graph Support for Gelly

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2564#discussion_r81218490
  
--- Diff: 
flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/BipartiteGraph.java
 ---
@@ -0,0 +1,231 @@
+/*
+ * 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.graph;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+/**
+ *
+ * Bipartite graph is a graph that contains two sets of vertices: top 
vertices and bottom vertices. Edges can only exist
+ * between a pair of vertices from different vertices sets. E.g. there can 
be no vertices between a pair
+ * of top vertices.
+ *
+ * Bipartite graph is useful to represent graphs with two sets of 
objects, like researchers and their publications,
+ * where an edge represents that a particular publication was authored by 
a particular author.
+ *
+ * Bipartite interface is different from {@link Graph} interface, so to 
apply algorithms that work on a regular graph
+ * a bipartite graph should be first converted into a {@link Graph} 
instance. This can be achieved by using
+ * {@link BipartiteGraph#topProjection(GroupReduceFunction)} or
+ * {@link BipartiteGraph#bottomProjection(GroupReduceFunction)} methods.
+ *
+ * @param  the key type of the top vertices
+ * @param  the key type of the bottom vertices
+ * @param  the top vertices value type
+ * @param  the bottom vertices value type
+ * @param  the edge value type
+ */
+public class BipartiteGraph {
+   private final ExecutionEnvironment context;
+   private final DataSet> topVertices;
+   private final DataSet> bottomVertices;
+   private final DataSet> edges;
+
+   private BipartiteGraph(
+   DataSet> topVertices,
+   DataSet> bottomVertices,
+   DataSet> edges,
+   ExecutionEnvironment context) {
+   this.topVertices = topVertices;
+   this.bottomVertices = bottomVertices;
+   this.edges = edges;
+   this.context = context;
+   }
+
+   /**
+* Create bipartite graph from datasets.
+*
+* @param topVertices  dataset of top vertices in the graph
+* @param bottomVertices dataset of bottom vertices in the graph
+* @param edges dataset of edges between vertices
+* @param context Flink execution context
+* @param  the key type of the top vertices
+* @param  the key type of the bottom vertices
+* @param  the top vertices value type
+* @param  the bottom vertices value type
+* @param  the edge value type
+* @return new bipartite graph created from provided datasets
+*/
+   public static  BipartiteGraph 
fromDataSet(
+   DataSet> topVertices,
+   DataSet> bottomVertices,
+   DataSet> edges,
+   ExecutionEnvironment context) {
+   return new BipartiteGraph<>(topVertices, bottomVertices, edges, 
context);
+   }
+
+   /**
+* Get dataset with top vertices.
+*
+* @return dataset with top vertices
+*/
+   public DataSet> getTopVertices() {
+   

[jira] [Commented] (FLINK-4329) Fix Streaming File Source Timestamps/Watermarks Handling

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2546#discussion_r81204828
  
--- Diff: 
flink-fs-tests/src/test/java/org/apache/flink/hdfstests/ContinuousFileMonitoringTest.java
 ---
@@ -440,10 +491,50 @@ public void testFileSplitMonitoringProcessOnce() 
throws Exception {
 
private int getLineNo(String line) {
String[] tkns = line.split("\\s");
-   Assert.assertEquals(tkns.length, 6);
+   Assert.assertEquals(6, tkns.length);
return Integer.parseInt(tkns[tkns.length - 1]);
}
 
+   private class TimeUpdatingThread extends Thread {
+
+   private volatile boolean isRunning;
+
+   private final TestTimeServiceProvider timeServiceProvider;
+   private final OneInputStreamOperatorTestHarness testHarness;
+   private final long wmInterval;
+   private final int elementUntilUpdating;
+
+   TimeUpdatingThread(final TestTimeServiceProvider 
timeServiceProvider,
+  final 
OneInputStreamOperatorTestHarness testHarness,
+  final long wmInterval,
+  final int 
elementUntilUpdating) {
+
+   this.timeServiceProvider = timeServiceProvider;
+   this.testHarness = testHarness;
+   this.wmInterval = wmInterval;
+   this.elementUntilUpdating = elementUntilUpdating;
+   this.isRunning = true;
+   }
+
+   @Override
+   public void run() {
+   try {
+   while (isRunning) {
+   if (testHarness.getOutput().size() % 
elementUntilUpdating == 0) {
+   long now = 
timeServiceProvider.getCurrentProcessingTime();
+   
timeServiceProvider.setCurrentTime(now + wmInterval);
+   }
+   }
+   } catch (Exception e) {
+   e.printStackTrace();
--- End diff --

This will not result in any meaningful feedback to the test.


> Fix Streaming File Source Timestamps/Watermarks Handling
> 
>
> Key: FLINK-4329
> URL: https://issues.apache.org/jira/browse/FLINK-4329
> Project: Flink
>  Issue Type: Bug
>  Components: Streaming Connectors
>Affects Versions: 1.1.0
>Reporter: Aljoscha Krettek
>Assignee: Kostas Kloudas
> Fix For: 1.2.0, 1.1.3
>
>
> The {{ContinuousFileReaderOperator}} does not correctly deal with watermarks, 
> i.e. they are just passed through. This means that when the 
> {{ContinuousFileMonitoringFunction}} closes and emits a {{Long.MAX_VALUE}} 
> that watermark can "overtake" the records that are to be emitted in the 
> {{ContinuousFileReaderOperator}}. Together with the new "allowed lateness" 
> setting in window operator this can lead to elements being dropped as late.
> Also, {{ContinuousFileReaderOperator}} does not correctly assign ingestion 
> timestamps since it is not technically a source but looks like one to the 
> user.



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

[GitHub] flink pull request #2546: [FLINK-4329] Fix Streaming File Source Timestamps/...

[StephanEwen]

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

https://github.com/apache/flink/pull/2546#discussion_r81204726
  
--- Diff: 
flink-fs-tests/src/test/java/org/apache/flink/hdfstests/ContinuousFileMonitoringTest.java
 ---
@@ -440,10 +491,50 @@ public void testFileSplitMonitoringProcessOnce() 
throws Exception {
 
private int getLineNo(String line) {
String[] tkns = line.split("\\s");
-   Assert.assertEquals(tkns.length, 6);
+   Assert.assertEquals(6, tkns.length);
return Integer.parseInt(tkns[tkns.length - 1]);
}
 
+   private class TimeUpdatingThread extends Thread {
+
+   private volatile boolean isRunning;
+
+   private final TestTimeServiceProvider timeServiceProvider;
+   private final OneInputStreamOperatorTestHarness testHarness;
+   private final long wmInterval;
+   private final int elementUntilUpdating;
+
+   TimeUpdatingThread(final TestTimeServiceProvider 
timeServiceProvider,
+  final 
OneInputStreamOperatorTestHarness testHarness,
+  final long wmInterval,
+  final int 
elementUntilUpdating) {
+
+   this.timeServiceProvider = timeServiceProvider;
+   this.testHarness = testHarness;
+   this.wmInterval = wmInterval;
+   this.elementUntilUpdating = elementUntilUpdating;
+   this.isRunning = true;
+   }
+
+   @Override
+   public void run() {
+   try {
+   while (isRunning) {
+   if (testHarness.getOutput().size() % 
elementUntilUpdating == 0) {
--- End diff --

There is a "race" between the operator emitting elements and this thread. 
Both run in loops without delays. Only if this condition is evaluated by chance 
at the exact point in time when the list happens to have so many result 
elements, there will actually be a time advance.


---
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-4329) Fix Streaming File Source Timestamps/Watermarks Handling

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2546#discussion_r81204990
  
--- Diff: 
flink-fs-tests/src/test/java/org/apache/flink/hdfstests/ContinuousFileMonitoringTest.java
 ---
@@ -190,12 +213,30 @@ public void 
testFileReadingOperatorWithIngestionTime() throws Exception {
}
content.add(element.getValue() + "\n");
} else if (line instanceof Watermark) {
-   Assert.assertEquals(((Watermark) 
line).getTimestamp(), Long.MAX_VALUE);
+   watermarkTimestamps.add(((Watermark) 
line).getTimestamp());
} else {
Assert.fail("Unknown element in the list.");
}
}
 
+   // check if all the input was read
+   Assert.assertEquals(NO_OF_FILES * LINES_PER_FILE, noOfLines);
+
+   // check if the last element is the LongMax watermark
+   Assert.assertTrue(lastElement instanceof Watermark);
+   Assert.assertEquals(Long.MAX_VALUE, ((Watermark) 
lastElement).getTimestamp());
+
+   System.out.println(watermarkTimestamps.size());
--- End diff --

Leftover sysout printing.


> Fix Streaming File Source Timestamps/Watermarks Handling
> 
>
> Key: FLINK-4329
> URL: https://issues.apache.org/jira/browse/FLINK-4329
> Project: Flink
>  Issue Type: Bug
>  Components: Streaming Connectors
>Affects Versions: 1.1.0
>Reporter: Aljoscha Krettek
>Assignee: Kostas Kloudas
> Fix For: 1.2.0, 1.1.3
>
>
> The {{ContinuousFileReaderOperator}} does not correctly deal with watermarks, 
> i.e. they are just passed through. This means that when the 
> {{ContinuousFileMonitoringFunction}} closes and emits a {{Long.MAX_VALUE}} 
> that watermark can "overtake" the records that are to be emitted in the 
> {{ContinuousFileReaderOperator}}. Together with the new "allowed lateness" 
> setting in window operator this can lead to elements being dropped as late.
> Also, {{ContinuousFileReaderOperator}} does not correctly assign ingestion 
> timestamps since it is not technically a source but looks like one to the 
> user.



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

[GitHub] flink pull request #2546: [FLINK-4329] Fix Streaming File Source Timestamps/...

[StephanEwen]

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

https://github.com/apache/flink/pull/2546#discussion_r81204990
  
--- Diff: 
flink-fs-tests/src/test/java/org/apache/flink/hdfstests/ContinuousFileMonitoringTest.java
 ---
@@ -190,12 +213,30 @@ public void 
testFileReadingOperatorWithIngestionTime() throws Exception {
}
content.add(element.getValue() + "\n");
} else if (line instanceof Watermark) {
-   Assert.assertEquals(((Watermark) 
line).getTimestamp(), Long.MAX_VALUE);
+   watermarkTimestamps.add(((Watermark) 
line).getTimestamp());
} else {
Assert.fail("Unknown element in the list.");
}
}
 
+   // check if all the input was read
+   Assert.assertEquals(NO_OF_FILES * LINES_PER_FILE, noOfLines);
+
+   // check if the last element is the LongMax watermark
+   Assert.assertTrue(lastElement instanceof Watermark);
+   Assert.assertEquals(Long.MAX_VALUE, ((Watermark) 
lastElement).getTimestamp());
+
+   System.out.println(watermarkTimestamps.size());
--- End diff --

Leftover sysout printing.


---
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 #2546: [FLINK-4329] Fix Streaming File Source Timestamps/...

[StephanEwen]

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

https://github.com/apache/flink/pull/2546#discussion_r81204828
  
--- Diff: 
flink-fs-tests/src/test/java/org/apache/flink/hdfstests/ContinuousFileMonitoringTest.java
 ---
@@ -440,10 +491,50 @@ public void testFileSplitMonitoringProcessOnce() 
throws Exception {
 
private int getLineNo(String line) {
String[] tkns = line.split("\\s");
-   Assert.assertEquals(tkns.length, 6);
+   Assert.assertEquals(6, tkns.length);
return Integer.parseInt(tkns[tkns.length - 1]);
}
 
+   private class TimeUpdatingThread extends Thread {
+
+   private volatile boolean isRunning;
+
+   private final TestTimeServiceProvider timeServiceProvider;
+   private final OneInputStreamOperatorTestHarness testHarness;
+   private final long wmInterval;
+   private final int elementUntilUpdating;
+
+   TimeUpdatingThread(final TestTimeServiceProvider 
timeServiceProvider,
+  final 
OneInputStreamOperatorTestHarness testHarness,
+  final long wmInterval,
+  final int 
elementUntilUpdating) {
+
+   this.timeServiceProvider = timeServiceProvider;
+   this.testHarness = testHarness;
+   this.wmInterval = wmInterval;
+   this.elementUntilUpdating = elementUntilUpdating;
+   this.isRunning = true;
+   }
+
+   @Override
+   public void run() {
+   try {
+   while (isRunning) {
+   if (testHarness.getOutput().size() % 
elementUntilUpdating == 0) {
+   long now = 
timeServiceProvider.getCurrentProcessingTime();
+   
timeServiceProvider.setCurrentTime(now + wmInterval);
+   }
+   }
+   } catch (Exception e) {
+   e.printStackTrace();
--- End diff --

This will not result in any meaningful feedback to the test.


---
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-4636) AbstractCEPPatternOperator fails to restore state

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2568#discussion_r81200110
  
--- Diff: 
flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPPatternOperator.java
 ---
@@ -126,11 +126,12 @@ public void restoreState(FSDataInputStream state) 
throws Exception {
 
int numberPriorityQueueEntries = ois.readInt();
 
-   priorityQueue = new 
PriorityQueue(numberPriorityQueueEntries, new 
StreamRecordComparator());
-
-   for (int i = 0; i  0) {
+   priorityQueue = new 
PriorityQueue(numberPriorityQueueEntries, new 
StreamRecordComparator());
--- End diff --

The queue is created in `open()`.


> AbstractCEPPatternOperator fails to restore state
> -
>
> Key: FLINK-4636
> URL: https://issues.apache.org/jira/browse/FLINK-4636
> Project: Flink
>  Issue Type: Bug
>  Components: CEP
>Affects Versions: 1.2.0, 1.1.2
>Reporter: Fabian Hueske
>Assignee: Jagadish Bihani
> Fix For: 1.2.0, 1.1.3
>
>
> The {{restoreState()}} of the {{AbstractCEPPatternOperator}} restores the a 
> Java {{PriorityQueue}}. For that it first reads the number of elements to 
> insert and then creates a {{PriorityQueue}} object. However, Java's 
> {{PriorityQueue}} cannot be instantiated with an initial capacity of {{0}}, 
> which is not checked.
> In case of an empty queue, the {{PriorityQueue}} should be instantiated with 
> an initial size of {{1}}.
> See 
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Problem-with-CEPPatternOperator-when-taskmanager-is-killed-tp9024.html



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

[jira] [Commented] (FLINK-4643) Average Clustering Coefficient

[ASF GitHub Bot (JIRA)]

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

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

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/2528
  
Looks good to me. Test failure is unrelated.

From my side, feel free to merge...



> Average Clustering Coefficient
> --
>
> Key: FLINK-4643
> URL: https://issues.apache.org/jira/browse/FLINK-4643
> Project: Flink
>  Issue Type: New Feature
>  Components: Gelly
>Affects Versions: 1.2.0
>Reporter: Greg Hogan
>Assignee: Greg Hogan
>Priority: Minor
>
> Gelly has Global Clustering Coefficient and Local Clustering Coefficient. 
> This adds Average Clustering Coefficient. The distinction is discussed in 
> [http://jponnela.com/web_documents/twomode.pdf] (pdf page 2, document page 
> 32).



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

[GitHub] flink pull request #2568: [FLINK-4636] Add boundary check for priorityqueue ...

[greghogan]

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

https://github.com/apache/flink/pull/2568#discussion_r81200110
  
--- Diff: 
flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPPatternOperator.java
 ---
@@ -126,11 +126,12 @@ public void restoreState(FSDataInputStream state) 
throws Exception {
 
int numberPriorityQueueEntries = ois.readInt();
 
-   priorityQueue = new 
PriorityQueue(numberPriorityQueueEntries, new 
StreamRecordComparator());
-
-   for (int i = 0; i  0) {
+   priorityQueue = new 
PriorityQueue(numberPriorityQueueEntries, new 
StreamRecordComparator());
--- End diff --

The queue is created in `open()`.


---
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 #2528: [FLINK-4643] [gelly] Average Clustering Coefficient

[StephanEwen]

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/2528
  
Looks good to me. Test failure is unrelated.

From my side, feel free to 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 pull request #2568: [FLINK-4636] Add boundary check for priorityqueue ...

[jaxbihani]

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

https://github.com/apache/flink/pull/2568#discussion_r81199485
  
--- Diff: 
flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPPatternOperator.java
 ---
@@ -126,11 +126,12 @@ public void restoreState(FSDataInputStream state) 
throws Exception {
 
int numberPriorityQueueEntries = ois.readInt();
 
-   priorityQueue = new 
PriorityQueue(numberPriorityQueueEntries, new 
StreamRecordComparator());
-
-   for (int i = 0; i  0) {
+   priorityQueue = new 
PriorityQueue(numberPriorityQueueEntries, new 
StreamRecordComparator());
--- End diff --

I think if number of objects read are 0 then why would we want to create 
object at all? Thats unnecessary and condition check will save some CPU cycles.


---
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-4636) AbstractCEPPatternOperator fails to restore state

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2568#discussion_r81199485
  
--- Diff: 
flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPPatternOperator.java
 ---
@@ -126,11 +126,12 @@ public void restoreState(FSDataInputStream state) 
throws Exception {
 
int numberPriorityQueueEntries = ois.readInt();
 
-   priorityQueue = new 
PriorityQueue(numberPriorityQueueEntries, new 
StreamRecordComparator());
-
-   for (int i = 0; i  0) {
+   priorityQueue = new 
PriorityQueue(numberPriorityQueueEntries, new 
StreamRecordComparator());
--- End diff --

I think if number of objects read are 0 then why would we want to create 
object at all? Thats unnecessary and condition check will save some CPU cycles.


> AbstractCEPPatternOperator fails to restore state
> -
>
> Key: FLINK-4636
> URL: https://issues.apache.org/jira/browse/FLINK-4636
> Project: Flink
>  Issue Type: Bug
>  Components: CEP
>Affects Versions: 1.2.0, 1.1.2
>Reporter: Fabian Hueske
>Assignee: Jagadish Bihani
> Fix For: 1.2.0, 1.1.3
>
>
> The {{restoreState()}} of the {{AbstractCEPPatternOperator}} restores the a 
> Java {{PriorityQueue}}. For that it first reads the number of elements to 
> insert and then creates a {{PriorityQueue}} object. However, Java's 
> {{PriorityQueue}} cannot be instantiated with an initial capacity of {{0}}, 
> which is not checked.
> In case of an empty queue, the {{PriorityQueue}} should be instantiated with 
> an initial size of {{1}}.
> See 
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Problem-with-CEPPatternOperator-when-taskmanager-is-killed-tp9024.html



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

[jira] [Created] (FLINK-4717) Naive version of atomic stop signal with savepoint

[Till Rohrmann (JIRA)]

Till Rohrmann created FLINK-4717:


 Summary: Naive version of atomic stop signal with savepoint
 Key: FLINK-4717
 URL: https://issues.apache.org/jira/browse/FLINK-4717
 Project: Flink
  Issue Type: New Feature
  Components: State Backends, Checkpointing
Affects Versions: 1.2.0
Reporter: Till Rohrmann
Priority: Minor
 Fix For: 1.2.0


As a first step towards atomic stopping with savepoints we should implement a 
cancel command which prior to cancelling takes a savepoint. Additionally, it 
should turn off the periodic checkpointing so that there won't be checkpoints 
executed between the savepoint and the cancel command.



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

[jira] [Commented] (FLINK-4573) Potential resource leak due to unclosed RandomAccessFile in TaskManagerLogHandler

[ASF GitHub Bot (JIRA)]

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

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

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/2556
  
Merging this...


> Potential resource leak due to unclosed RandomAccessFile in 
> TaskManagerLogHandler
> -
>
> Key: FLINK-4573
> URL: https://issues.apache.org/jira/browse/FLINK-4573
> Project: Flink
>  Issue Type: Bug
>Reporter: Ted Yu
>Priority: Minor
>
> {code}
> try {
> raf = new 
> RandomAccessFile(file, "r");
> } catch 
> (FileNotFoundException e) {
> display(ctx, request, 
> "Displaying TaskManager log failed.");
> LOG.error("Displaying 
> TaskManager log failed.", e);
> return;
> }
> long fileLength = 
> raf.length();
> final FileChannel fc = 
> raf.getChannel();
> {code}
> If length() throws IOException, raf would be left unclosed.



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

[GitHub] flink issue #2556: [FLINK-4573] Fix potential resource leak due to unclosed ...

[StephanEwen]

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/2556
  
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.
---

[jira] [Updated] (FLINK-4650) Frequent task manager disconnects from JobManager

[Till Rohrmann (JIRA)]

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

Till Rohrmann updated FLINK-4650:
-
Fix Version/s: 1.2.0

> Frequent task manager disconnects from JobManager
> -
>
> Key: FLINK-4650
> URL: https://issues.apache.org/jira/browse/FLINK-4650
> Project: Flink
>  Issue Type: Bug
>  Components: Distributed Coordination, Network
>Affects Versions: 1.2.0
>Reporter: Nagarjun Guraja
> Fix For: 1.2.0
>
>
> Not sure of the exact reason but we observe more frequent task manager 
> disconnects while using 1.2 snapshot build as compared to 1.1.2 release build



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

[jira] [Updated] (FLINK-4650) Frequent task manager disconnects from JobManager

[Till Rohrmann (JIRA)]

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

Till Rohrmann updated FLINK-4650:
-
Affects Version/s: 1.2.0

> Frequent task manager disconnects from JobManager
> -
>
> Key: FLINK-4650
> URL: https://issues.apache.org/jira/browse/FLINK-4650
> Project: Flink
>  Issue Type: Bug
>  Components: Distributed Coordination, Network
>Affects Versions: 1.2.0
>Reporter: Nagarjun Guraja
> Fix For: 1.2.0
>
>
> Not sure of the exact reason but we observe more frequent task manager 
> disconnects while using 1.2 snapshot build as compared to 1.1.2 release build



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

[jira] [Created] (FLINK-4716) Add trigger full recovery button to web UI

[Till Rohrmann (JIRA)]

Till Rohrmann created FLINK-4716:


 Summary: Add trigger full recovery button to web UI
 Key: FLINK-4716
 URL: https://issues.apache.org/jira/browse/FLINK-4716
 Project: Flink
  Issue Type: Improvement
  Components: Webfrontend
Affects Versions: 1.2.0
Reporter: Till Rohrmann
Priority: Minor
 Fix For: 1.2.0


Add a trigger full recovery button to the web UI. The full recovery button will 
take the latest completed checkpoint and resume from there.



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

[jira] [Created] (FLINK-4715) TaskManager should commit suicide after cancellation failure

[Till Rohrmann (JIRA)]

Till Rohrmann created FLINK-4715:


 Summary: TaskManager should commit suicide after cancellation 
failure
 Key: FLINK-4715
 URL: https://issues.apache.org/jira/browse/FLINK-4715
 Project: Flink
  Issue Type: Improvement
  Components: TaskManager
Affects Versions: 1.2.0
Reporter: Till Rohrmann
 Fix For: 1.2.0


In case of a failed cancellation, e.g. the task cannot be cancelled after a 
given time, the {{TaskManager}} should kill itself. That way we guarantee that 
there is no resource leak. 

This behaviour acts as a safety-net against faulty user code.



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

[jira] [Created] (FLINK-4714) Set task state to RUNNING after state has been restored

[Till Rohrmann (JIRA)]

Till Rohrmann created FLINK-4714:


 Summary: Set task state to RUNNING after state has been restored
 Key: FLINK-4714
 URL: https://issues.apache.org/jira/browse/FLINK-4714
 Project: Flink
  Issue Type: Improvement
  Components: Distributed Coordination, State Backends, Checkpointing
Affects Versions: 1.2.0
Reporter: Till Rohrmann
 Fix For: 1.2.0


The task state is set to {{RUNNING}} as soon as the {{Task}} is executed. That, 
however, happens before the state of the {{StreamTask}} invokable has been 
restored. As a result, the {{CheckpointCoordinator}} starts to trigger 
checkpoints even though the {{StreamTask}} is not ready.

In order to avoid aborting checkpoints and properly start it, we should switch 
the task state to {{RUNNING}} after the state has been restored.



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

[jira] [Commented] (FLINK-4700) Harden the TimeProvider test

[Kostas Kloudas (JIRA)]

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

Kostas Kloudas commented on FLINK-4700:
---

[~StephanEwen] 

I am thinking that the best solution to this is to remove the 
testDefaultTimeProvider() test altogether.

The only thing it actually tests is the java ScheduledExecutorService and there 
is no way to make it stable 
without adding test-specific methods (visible for testing) in the 
DefaultTimeServiceProvider. 

What do you think?

> Harden the TimeProvider test
> 
>
> Key: FLINK-4700
> URL: https://issues.apache.org/jira/browse/FLINK-4700
> Project: Flink
>  Issue Type: Bug
>  Components: Tests
>Reporter: Kostas Kloudas
>Assignee: Kostas Kloudas
>
> Currently the TimeProvider test fails due to a race condition. This task aims 
> at fixing it.



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

[jira] [Commented] (FLINK-4702) Kafka consumer must commit offsets asynchronously

[ASF GitHub Bot (JIRA)]

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

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

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/2574
  
@tzulitai @rmetzger We need to make sure that the Kafka 0.10 code picks up 
this change and the test case.


> Kafka consumer must commit offsets asynchronously
> -
>
> Key: FLINK-4702
> URL: https://issues.apache.org/jira/browse/FLINK-4702
> Project: Flink
>  Issue Type: Bug
>  Components: Kafka Connector
>Affects Versions: 1.1.2
>Reporter: Stephan Ewen
>Assignee: Stephan Ewen
>Priority: Blocker
> Fix For: 1.2.0, 1.1.3
>
>
> The offset commit calls to Kafka may occasionally take very long.
> In that case, the {{notifyCheckpointComplete()}} method blocks for long and 
> the KafkaConsumer cannot make progress and cannot perform checkpoints.
> Kafka 0.9+ have methods to commit asynchronously.
> We should use those and make sure no more than one commit is concurrently in 
> progress, to that commit requests do not pile up.



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

[jira] [Commented] (FLINK-4711) TaskManager can crash due to failing onPartitionStateUpdate call

[ASF GitHub Bot (JIRA)]

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

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

Github user tillrohrmann commented on the issue:

https://github.com/apache/flink/pull/2569
  
Thanks for the review @uce. If Travis gives green light, I'll merge the PR.


> TaskManager can crash due to failing onPartitionStateUpdate call
> 
>
> Key: FLINK-4711
> URL: https://issues.apache.org/jira/browse/FLINK-4711
> Project: Flink
>  Issue Type: Bug
>  Components: Distributed Coordination
>Affects Versions: 1.2.0
>Reporter: Till Rohrmann
>Assignee: Till Rohrmann
> Fix For: 1.2.0
>
>
> The {{TaskManager}} can crash because it calls 
> {{Task.onPartitionStateUpdate}} when it receives a {{PartitionState}} 
> message. The {{onPartitionStateUpdate}} method can throw an {{IOException}} 
> or {{InterruptedException}} which are not handled on the {{TaskManager}} 
> level.
> Another problem is that the initial partition state request is triggered 
> within the {{SingleInputGate}}. The request causes the {{JobManager}} to send 
> a {{PartitionState}} message to the {{TaskManager}} which forwards it to the 
> {{Task}}. If the at any of these points a message gets lost, then it is not 
> retried and the partition state remains unknown.
> In order to handle the exceptions, to make the data flow clearer and to add 
> automatic retries, I propose to let the {{Task}} send the partition state 
> check requests. Furthermore, the {{JobManager}} should directly answer to the 
> {{Task}} by replying to an ask operation. That way the message does not have 
> to be routed through the {{TaskManager}}.



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

[GitHub] flink issue #2574: [FLINK-4702] [kafka connector] Commit offsets to Kafka as...

[StephanEwen]

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/2574
  
@tzulitai @rmetzger We need to make sure that the Kafka 0.10 code picks up 
this change and the test case.


---
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 #2569: [FLINK-4711] Let the Task trigger partition state request...

[tillrohrmann]

Github user tillrohrmann commented on the issue:

https://github.com/apache/flink/pull/2569
  
Thanks for the review @uce. If Travis gives green light, I'll merge the PR.


---
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-4702) Kafka consumer must commit offsets asynchronously

[ASF GitHub Bot (JIRA)]

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

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

Github user StephanEwen closed the pull request at:

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


> Kafka consumer must commit offsets asynchronously
> -
>
> Key: FLINK-4702
> URL: https://issues.apache.org/jira/browse/FLINK-4702
> Project: Flink
>  Issue Type: Bug
>  Components: Kafka Connector
>Affects Versions: 1.1.2
>Reporter: Stephan Ewen
>Assignee: Stephan Ewen
>Priority: Blocker
> Fix For: 1.2.0, 1.1.3
>
>
> The offset commit calls to Kafka may occasionally take very long.
> In that case, the {{notifyCheckpointComplete()}} method blocks for long and 
> the KafkaConsumer cannot make progress and cannot perform checkpoints.
> Kafka 0.9+ have methods to commit asynchronously.
> We should use those and make sure no more than one commit is concurrently in 
> progress, to that commit requests do not pile up.



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

[jira] [Commented] (FLINK-4702) Kafka consumer must commit offsets asynchronously

[ASF GitHub Bot (JIRA)]

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

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

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/2559
  
Closing this for #2574 


> Kafka consumer must commit offsets asynchronously
> -
>
> Key: FLINK-4702
> URL: https://issues.apache.org/jira/browse/FLINK-4702
> Project: Flink
>  Issue Type: Bug
>  Components: Kafka Connector
>Affects Versions: 1.1.2
>Reporter: Stephan Ewen
>Assignee: Stephan Ewen
>Priority: Blocker
> Fix For: 1.2.0, 1.1.3
>
>
> The offset commit calls to Kafka may occasionally take very long.
> In that case, the {{notifyCheckpointComplete()}} method blocks for long and 
> the KafkaConsumer cannot make progress and cannot perform checkpoints.
> Kafka 0.9+ have methods to commit asynchronously.
> We should use those and make sure no more than one commit is concurrently in 
> progress, to that commit requests do not pile up.



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

[GitHub] flink pull request #2559: [FLINK-4702] [kafka connector] Commit offets to Ka...

[StephanEwen]

Github user StephanEwen closed the pull request at:

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


---
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 #2559: [FLINK-4702] [kafka connector] Commit offets to Kafka asy...

[StephanEwen]

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/2559
  
Closing this for #2574 


---
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-4702) Kafka consumer must commit offsets asynchronously

[ASF GitHub Bot (JIRA)]

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

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

GitHub user StephanEwen opened a pull request:

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

[FLINK-4702] [kafka connector] Commit offsets to Kafka asynchronously and 
don't block on polls

This fix is quite critical!

While the KafkaConsumer is polling for new data (with a timeout), it holds 
the consumer lock. If no data comes in Kafka, the lock is not released before 
the poll timeout is over.

During that time, no offset commit can make progress, because it needs the 
consumer lock. The `notifyCheckpointComplete()` method of the Kafka Consumer 
hence blocks until the poll timeout is over and the lock is released. For 
low-throughput Kafka Topics, this can cause wildly long checkpoint delays.

This changes `notifyCheckpointComplete()` to only "schedule" offsets to be 
committed, while the main fetcher thread actually kick off the asynchronous 
offset commits. That way, there is no interference between the 
`notifyCheckpointComplete()` method (which is executed under checkpoint lock) 
and the consumer lock.

In fact, the only KafkaConsumer method accessed concurrently to the main 
fetcher thread is `wakeup()` which is actually thread-safe (where the rest of 
the KafkaConsumer is not). The consumer lock was hence completely removed.


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

$ git pull https://github.com/StephanEwen/incubator-flink kafka_09_fix

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

https://github.com/apache/flink/pull/2574.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 #2574


commit 0846fd907db7d52d7e5fb7d704c5e1c13462e331
Author: Stephan Ewen 
Date:   2016-09-29T16:09:51Z

[FLINK-4702] [kafka connector] Commit offsets to Kafka asynchronously and 
don't block on polls

Letting the Kafka commit block on polls means that 
'notifyCheckpointComplete()' may take
very long. This is mostly relevant for low-throughput Kafka topics.




> Kafka consumer must commit offsets asynchronously
> -
>
> Key: FLINK-4702
> URL: https://issues.apache.org/jira/browse/FLINK-4702
> Project: Flink
>  Issue Type: Bug
>  Components: Kafka Connector
>Affects Versions: 1.1.2
>Reporter: Stephan Ewen
>Assignee: Stephan Ewen
>Priority: Blocker
> Fix For: 1.2.0, 1.1.3
>
>
> The offset commit calls to Kafka may occasionally take very long.
> In that case, the {{notifyCheckpointComplete()}} method blocks for long and 
> the KafkaConsumer cannot make progress and cannot perform checkpoints.
> Kafka 0.9+ have methods to commit asynchronously.
> We should use those and make sure no more than one commit is concurrently in 
> progress, to that commit requests do not pile up.



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

[jira] [Commented] (FLINK-4348) Implement slot allocation protocol with TaskExecutor

[ASF GitHub Bot (JIRA)]

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

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

Github user mxm commented on the issue:

https://github.com/apache/flink/pull/2571
  
CC @beyond1920 @KurtYoung 


> Implement slot allocation protocol with TaskExecutor
> 
>
> Key: FLINK-4348
> URL: https://issues.apache.org/jira/browse/FLINK-4348
> Project: Flink
>  Issue Type: Sub-task
>  Components: Cluster Management
>Reporter: Kurt Young
>Assignee: Maximilian Michels
>
> When slotManager finds a proper slot in the free pool for a slot request,  
> slotManager marks the slot as occupied, then tells the taskExecutor to give 
> the slot to the specified JobMaster. 
> when a slot request is sent to taskExecutor, it should contain following 
> parameters: AllocationID, JobID,  slotID, resourceManagerLeaderSessionID. 
> There exists 3 following possibilities of the response from taskExecutor, we 
> will discuss when each possibility happens and how to handle.
> 1. Ack request which means the taskExecutor gives the slot to the specified 
> jobMaster as expected.   
> 2. Decline request if the slot is already occupied by other AllocationID.  
> 3. Timeout which could caused by lost of request message or response message 
> or slow network transfer. 
> On the first occasion, ResourceManager need to do nothing. However, under the 
> second and third occasion, ResourceManager need to notify slotManager, 
> slotManager will verify and clear all the previous allocate information for 
> this slot request firstly, then try to find a proper slot for the slot 
> request again. This may cause some duplicate allocation, e.g. the slot 
> request to TaskManager is successful but the response is lost somehow, so we 
> may request a slot in another TaskManager, this causes two slots assigned to 
> one request, but it can be taken care of by rejecting registration at 
> JobMaster.
> There are still some question need to discuss in a step further.
> 1. Who send slotRequest to taskExecutor, SlotManager or ResourceManager? I 
> think it's better that SlotManager delegates the rpc call to ResourceManager 
> when SlotManager need to communicate with outside world.  ResourceManager 
> know which taskExecutor to send the request based on ResourceID. Besides this 
> RPC call which used to request slot to taskExecutor should not be a 
> RpcMethod,  because we hope only SlotManager has permission to call the 
> method, but the other component, for example JobMaster and TaskExecutor, 
> cannot call this method directly.
> 2. If JobMaster reject the slot offer from a TaskExecutor, the TaskExecutor 
> should notify the free slot to ResourceManager immediately, or wait for next 
> heartbeat sync. The advantage of first way is the resourceManager’s view 
> could be updated faster. The advantage of second way is save a RPC method in 
> ResourceManager.
> 3. There are two communication type. First, the slot request could be sent as 
> an ask operation where the response is returned as a future. Second, 
> resourceManager send the slot request in fire and forget way, the response 
> could be returned by an RPC call. I prefer the first one because it is more 
> simple and could save a RPC method in ResourceManager (for callback in the 
> second way).



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

[GitHub] flink pull request #2574: [FLINK-4702] [kafka connector] Commit offsets to K...

[StephanEwen]

GitHub user StephanEwen opened a pull request:

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

[FLINK-4702] [kafka connector] Commit offsets to Kafka asynchronously and 
don't block on polls

This fix is quite critical!

While the KafkaConsumer is polling for new data (with a timeout), it holds 
the consumer lock. If no data comes in Kafka, the lock is not released before 
the poll timeout is over.

During that time, no offset commit can make progress, because it needs the 
consumer lock. The `notifyCheckpointComplete()` method of the Kafka Consumer 
hence blocks until the poll timeout is over and the lock is released. For 
low-throughput Kafka Topics, this can cause wildly long checkpoint delays.

This changes `notifyCheckpointComplete()` to only "schedule" offsets to be 
committed, while the main fetcher thread actually kick off the asynchronous 
offset commits. That way, there is no interference between the 
`notifyCheckpointComplete()` method (which is executed under checkpoint lock) 
and the consumer lock.

In fact, the only KafkaConsumer method accessed concurrently to the main 
fetcher thread is `wakeup()` which is actually thread-safe (where the rest of 
the KafkaConsumer is not). The consumer lock was hence completely removed.


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

$ git pull https://github.com/StephanEwen/incubator-flink kafka_09_fix

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

https://github.com/apache/flink/pull/2574.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 #2574


commit 0846fd907db7d52d7e5fb7d704c5e1c13462e331
Author: Stephan Ewen 
Date:   2016-09-29T16:09:51Z

[FLINK-4702] [kafka connector] Commit offsets to Kafka asynchronously and 
don't block on polls

Letting the Kafka commit block on polls means that 
'notifyCheckpointComplete()' may take
very long. This is mostly relevant for low-throughput Kafka topics.




---
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 #2571: [FLINK-4348] Simplify logic of SlotManager

[mxm]

Github user mxm commented on the issue:

https://github.com/apache/flink/pull/2571
  
CC @beyond1920 @KurtYoung 


---
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-4711) TaskManager can crash due to failing onPartitionStateUpdate call

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2569#discussion_r81173562
  
--- Diff: 
flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskControlMessages.scala
 ---
@@ -92,16 +92,6 @@ object TaskMessages {
   // 
--
 
   /**
-   * Answer to a [[RequestPartitionState]] with the state of the 
respective partition.
-   */
-  case class PartitionState(
--- End diff --

Yes that is also true. Will fix it.


> TaskManager can crash due to failing onPartitionStateUpdate call
> 
>
> Key: FLINK-4711
> URL: https://issues.apache.org/jira/browse/FLINK-4711
> Project: Flink
>  Issue Type: Bug
>  Components: Distributed Coordination
>Affects Versions: 1.2.0
>Reporter: Till Rohrmann
>Assignee: Till Rohrmann
> Fix For: 1.2.0
>
>
> The {{TaskManager}} can crash because it calls 
> {{Task.onPartitionStateUpdate}} when it receives a {{PartitionState}} 
> message. The {{onPartitionStateUpdate}} method can throw an {{IOException}} 
> or {{InterruptedException}} which are not handled on the {{TaskManager}} 
> level.
> Another problem is that the initial partition state request is triggered 
> within the {{SingleInputGate}}. The request causes the {{JobManager}} to send 
> a {{PartitionState}} message to the {{TaskManager}} which forwards it to the 
> {{Task}}. If the at any of these points a message gets lost, then it is not 
> retried and the partition state remains unknown.
> In order to handle the exceptions, to make the data flow clearer and to add 
> automatic retries, I propose to let the {{Task}} send the partition state 
> check requests. Furthermore, the {{JobManager}} should directly answer to the 
> {{Task}} by replying to an ask operation. That way the message does not have 
> to be routed through the {{TaskManager}}.



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

[jira] [Commented] (FLINK-4711) TaskManager can crash due to failing onPartitionStateUpdate call

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2569#discussion_r81173599
  
--- Diff: 
flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
 ---
@@ -873,8 +873,7 @@ class JobManager(
   }
 
   sender ! decorateMessage(
-PartitionState(
-  taskExecutionId,
+new org.apache.flink.runtime.io.network.PartitionState(
--- End diff --

Yes that's true. Will fix it.


> TaskManager can crash due to failing onPartitionStateUpdate call
> 
>
> Key: FLINK-4711
> URL: https://issues.apache.org/jira/browse/FLINK-4711
> Project: Flink
>  Issue Type: Bug
>  Components: Distributed Coordination
>Affects Versions: 1.2.0
>Reporter: Till Rohrmann
>Assignee: Till Rohrmann
> Fix For: 1.2.0
>
>
> The {{TaskManager}} can crash because it calls 
> {{Task.onPartitionStateUpdate}} when it receives a {{PartitionState}} 
> message. The {{onPartitionStateUpdate}} method can throw an {{IOException}} 
> or {{InterruptedException}} which are not handled on the {{TaskManager}} 
> level.
> Another problem is that the initial partition state request is triggered 
> within the {{SingleInputGate}}. The request causes the {{JobManager}} to send 
> a {{PartitionState}} message to the {{TaskManager}} which forwards it to the 
> {{Task}}. If the at any of these points a message gets lost, then it is not 
> retried and the partition state remains unknown.
> In order to handle the exceptions, to make the data flow clearer and to add 
> automatic retries, I propose to let the {{Task}} send the partition state 
> check requests. Furthermore, the {{JobManager}} should directly answer to the 
> {{Task}} by replying to an ask operation. That way the message does not have 
> to be routed through the {{TaskManager}}.



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

[GitHub] flink pull request #2569: [FLINK-4711] Let the Task trigger partition state ...

[tillrohrmann]

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

https://github.com/apache/flink/pull/2569#discussion_r81173562
  
--- Diff: 
flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskControlMessages.scala
 ---
@@ -92,16 +92,6 @@ object TaskMessages {
   // 
--
 
   /**
-   * Answer to a [[RequestPartitionState]] with the state of the 
respective partition.
-   */
-  case class PartitionState(
--- End diff --

Yes that is also true. Will fix it.


---
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 #2569: [FLINK-4711] Let the Task trigger partition state ...

[tillrohrmann]

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

https://github.com/apache/flink/pull/2569#discussion_r81173599
  
--- Diff: 
flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
 ---
@@ -873,8 +873,7 @@ class JobManager(
   }
 
   sender ! decorateMessage(
-PartitionState(
-  taskExecutionId,
+new org.apache.flink.runtime.io.network.PartitionState(
--- End diff --

Yes that's true. Will fix it.


---
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-4711) TaskManager can crash due to failing onPartitionStateUpdate call

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2569#discussion_r81173481
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/JobManagerCommunicationFactory.java
 ---
@@ -27,15 +26,6 @@
 public interface JobManagerCommunicationFactory {
--- End diff --

Good catch. Will remove it.


> TaskManager can crash due to failing onPartitionStateUpdate call
> 
>
> Key: FLINK-4711
> URL: https://issues.apache.org/jira/browse/FLINK-4711
> Project: Flink
>  Issue Type: Bug
>  Components: Distributed Coordination
>Affects Versions: 1.2.0
>Reporter: Till Rohrmann
>Assignee: Till Rohrmann
> Fix For: 1.2.0
>
>
> The {{TaskManager}} can crash because it calls 
> {{Task.onPartitionStateUpdate}} when it receives a {{PartitionState}} 
> message. The {{onPartitionStateUpdate}} method can throw an {{IOException}} 
> or {{InterruptedException}} which are not handled on the {{TaskManager}} 
> level.
> Another problem is that the initial partition state request is triggered 
> within the {{SingleInputGate}}. The request causes the {{JobManager}} to send 
> a {{PartitionState}} message to the {{TaskManager}} which forwards it to the 
> {{Task}}. If the at any of these points a message gets lost, then it is not 
> retried and the partition state remains unknown.
> In order to handle the exceptions, to make the data flow clearer and to add 
> automatic retries, I propose to let the {{Task}} send the partition state 
> check requests. Furthermore, the {{JobManager}} should directly answer to the 
> {{Task}} by replying to an ask operation. That way the message does not have 
> to be routed through the {{TaskManager}}.



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

[GitHub] flink pull request #2569: [FLINK-4711] Let the Task trigger partition state ...

[tillrohrmann]

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

https://github.com/apache/flink/pull/2569#discussion_r81173481
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/JobManagerCommunicationFactory.java
 ---
@@ -27,15 +26,6 @@
 public interface JobManagerCommunicationFactory {
--- End diff --

Good catch. Will remove it.


---
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 #2573: Refactor StreamOperator Hierachy to make Keys Expl...

[aljoscha]

GitHub user aljoscha opened a pull request:

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

Refactor StreamOperator Hierachy to make Keys Explicit

This is more of a preview PR, I'm not yet completely done with testing and 
making sure that everything works.

R: @StephanEwen 

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

$ git pull https://github.com/aljoscha/flink refactor-stream-operators

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

https://github.com/apache/flink/pull/2573.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 #2573


commit c9049c57d612e18f6051574d9903bb063b46840a
Author: Aljoscha Krettek 
Date:   2016-09-28T09:07:22Z

Refactor StreamOperator Hierachy to make Keys Explicit




---
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-4552) Refactor WindowOperator/Trigger Tests

[ASF GitHub Bot (JIRA)]

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

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

GitHub user aljoscha opened a pull request:

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

[FLINK-4552] Refactor WindowOperator/Trigger Tests

This builds on #2570 

Before, tests for WindowOperator, WindowAssigner, Trigger and WindowFunction
were all conflated in WindowOperatorTest. All of these tested that a
certain combination of a Trigger, WindowAssigner and WindowFunction produce
the expected output.

This change modularizes these tests and spreads them out across multiple
files. For example, one per trigger/window assigner.

The new WindowOperatorTest tests verify that the interaction between
WindowOperator and the various other parts works as expected, that the
correct methods on Trigger and WindowFunction are called at the expected
time and that snapshotting, timers, cleanup etc. work correctly. These tests
also verify that the different state types and WindowFunctions work 
correctly.

For trigger tests this introduces TriggerTestHarness. This can be used
to inject elements into Triggers they fire at the correct times. The
actual output of the WindowFunction is not important for these tests.
The new tests also make sure that triggers correctly clean up state and 
timers.

WindowAssigner tests verify the behaviour of window assigners in isolation.
They also test, for example, whether offset parameter of time-based windows
work correctly.

R: @StephanEwen @StefanRRichter @kl0u for review

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

$ git pull https://github.com/aljoscha/flink window-test-refactor

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

https://github.com/apache/flink/pull/2572.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 #2572


commit 1a09d9032bf5683a378a7fc8dc480f2d14c5924d
Author: Aljoscha Krettek 
Date:   2016-09-25T18:58:16Z

Rename TimeServiceProvider to ProcessingTimeService

The name is clashing with the soon-to-be-added
TimerService/InternalTimerService which is meant as an interface for
dealing with both processing time and event time.

TimeServiceProvided is renamed to ProcessingTimeService to reflect the
fact that it is a low-level utility that only deals with "physical"
processing-time trigger tasks.

commit 758827c3c8508ef9ef2ec079ff3a8469d0096ca8
Author: Aljoscha Krettek 
Date:   2016-09-28T13:10:35Z

Use OperatorTestHarness and TestProcessingTimeService in Kafka Tests

commit f6dd9c74dc2c58c4263fb6d084651b514898d47a
Author: Aljoscha Krettek 
Date:   2016-09-28T14:35:33Z

Use Processing-Time Service of TestHarness in WindowOperatorTest

Before, this was manually creating a TestProcessingTimeService, now,
we're using the one that is there by default in
OneInputStreamOperatorTestHarness.

commit 65389d66c5586e6707b7a6bf48df512354fac085
Author: Aljoscha Krettek 
Date:   2016-09-28T14:43:40Z

Refactor OperatorTestHarness to always use TestProcessingTimeService

Before, this would allow handing in a custom ProcessingTimeService but
this was in reality always TestProcessingTimeService.

commit 1d013bcacc040552e5783c64d094ec309014457b
Author: Aljoscha Krettek 
Date:   2016-09-28T13:12:26Z

Use TestHarness Processing-time Facility in BucketingSinkTest

Before, this was manually creating a TestProcessingTimeService. Now we
use the one that is there by default in
OneInputStreamOperatorTestHarness.

commit eaf3dd00fefeb2487c7cafff6337123cbe42874b
Author: Aljoscha Krettek 
Date:   2016-09-28T13:32:24Z

Use OperatorTestHarness in AlignedWindowOperator Tests

commit b597d2ef50c27554b83fddaff8873107265340d4
Author: Aljoscha Krettek 
Date:   2016-09-29T14:04:29Z

Refactor Operator TestHarnesses to use Common Base Class

This also introduces KeyedTwoInputStreamOperatorTestHarness which
is similar to KeyedOneInputStreamOperatorTestHarness

commit 58b16b26e07b6100f89e9deec63f0decb751f0e6
Author: Aljoscha Krettek 
Date:   2016-09-26T14:21:51Z

[FLINK-3674] Add an interface for Time aware User Functions

This moves the event-time/processing-time trigger code from
WindowOperator behind a well defined interface that can be used by
operators (and user functions).

InternalTimerService is the new interface that has the same

[GitHub] flink pull request #2572: [FLINK-4552] Refactor WindowOperator/Trigger Tests

[aljoscha]

GitHub user aljoscha opened a pull request:

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

[FLINK-4552] Refactor WindowOperator/Trigger Tests

This builds on #2570 

Before, tests for WindowOperator, WindowAssigner, Trigger and WindowFunction
were all conflated in WindowOperatorTest. All of these tested that a
certain combination of a Trigger, WindowAssigner and WindowFunction produce
the expected output.

This change modularizes these tests and spreads them out across multiple
files. For example, one per trigger/window assigner.

The new WindowOperatorTest tests verify that the interaction between
WindowOperator and the various other parts works as expected, that the
correct methods on Trigger and WindowFunction are called at the expected
time and that snapshotting, timers, cleanup etc. work correctly. These tests
also verify that the different state types and WindowFunctions work 
correctly.

For trigger tests this introduces TriggerTestHarness. This can be used
to inject elements into Triggers they fire at the correct times. The
actual output of the WindowFunction is not important for these tests.
The new tests also make sure that triggers correctly clean up state and 
timers.

WindowAssigner tests verify the behaviour of window assigners in isolation.
They also test, for example, whether offset parameter of time-based windows
work correctly.

R: @StephanEwen @StefanRRichter @kl0u for review

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

$ git pull https://github.com/aljoscha/flink window-test-refactor

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

https://github.com/apache/flink/pull/2572.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 #2572


commit 1a09d9032bf5683a378a7fc8dc480f2d14c5924d
Author: Aljoscha Krettek 
Date:   2016-09-25T18:58:16Z

Rename TimeServiceProvider to ProcessingTimeService

The name is clashing with the soon-to-be-added
TimerService/InternalTimerService which is meant as an interface for
dealing with both processing time and event time.

TimeServiceProvided is renamed to ProcessingTimeService to reflect the
fact that it is a low-level utility that only deals with "physical"
processing-time trigger tasks.

commit 758827c3c8508ef9ef2ec079ff3a8469d0096ca8
Author: Aljoscha Krettek 
Date:   2016-09-28T13:10:35Z

Use OperatorTestHarness and TestProcessingTimeService in Kafka Tests

commit f6dd9c74dc2c58c4263fb6d084651b514898d47a
Author: Aljoscha Krettek 
Date:   2016-09-28T14:35:33Z

Use Processing-Time Service of TestHarness in WindowOperatorTest

Before, this was manually creating a TestProcessingTimeService, now,
we're using the one that is there by default in
OneInputStreamOperatorTestHarness.

commit 65389d66c5586e6707b7a6bf48df512354fac085
Author: Aljoscha Krettek 
Date:   2016-09-28T14:43:40Z

Refactor OperatorTestHarness to always use TestProcessingTimeService

Before, this would allow handing in a custom ProcessingTimeService but
this was in reality always TestProcessingTimeService.

commit 1d013bcacc040552e5783c64d094ec309014457b
Author: Aljoscha Krettek 
Date:   2016-09-28T13:12:26Z

Use TestHarness Processing-time Facility in BucketingSinkTest

Before, this was manually creating a TestProcessingTimeService. Now we
use the one that is there by default in
OneInputStreamOperatorTestHarness.

commit eaf3dd00fefeb2487c7cafff6337123cbe42874b
Author: Aljoscha Krettek 
Date:   2016-09-28T13:32:24Z

Use OperatorTestHarness in AlignedWindowOperator Tests

commit b597d2ef50c27554b83fddaff8873107265340d4
Author: Aljoscha Krettek 
Date:   2016-09-29T14:04:29Z

Refactor Operator TestHarnesses to use Common Base Class

This also introduces KeyedTwoInputStreamOperatorTestHarness which
is similar to KeyedOneInputStreamOperatorTestHarness

commit 58b16b26e07b6100f89e9deec63f0decb751f0e6
Author: Aljoscha Krettek 
Date:   2016-09-26T14:21:51Z

[FLINK-3674] Add an interface for Time aware User Functions

This moves the event-time/processing-time trigger code from
WindowOperator behind a well defined interface that can be used by
operators (and user functions).

InternalTimerService is the new interface that has the same
functionality that WindowOperator used to have. TimerService is the user
facing interface that does not allow dealing with namespaces/payloads
and also does not allow deleting timers. There is a default
implementation in 

[jira] [Commented] (FLINK-4348) Implement slot allocation protocol with TaskExecutor

[ASF GitHub Bot (JIRA)]

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

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

GitHub user mxm opened a pull request:

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

[FLINK-4348] Simplify logic of SlotManager

This pull request is split up into two commits:

1. It removes some code from the `SlotManager` to make it simpler. 
  - It makes use of `handleSlotRequestFailedAtTaskManager` and simplifies 
it because we can assume that a Slot is only registered once if we talk to the 
same instance of a TaskExecutor. Further, we can omit to check the free slots 
because we previously removed the slot from the free slots.
  - `updateSlotStatus` only deals with new Task slots. All other updates 
are performed by the `ResourceManager`. In case the ResourceManager creashes, 
it will re-register all slot statuses.

2. It fences `TaskExecutor` messages using an `InstanceID` which is 
required to make 1 work correctly. New messages have been introduced to achieve 
that.

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

$ git pull https://github.com/mxm/flink flip-6

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

https://github.com/apache/flink/pull/2571.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 #2571


commit 1d297403e5e085d944d80b498c4a269a794f8e2f
Author: Maximilian Michels 
Date:   2016-09-29T13:08:32Z

[FLINK-4347] change assumptions in SlotManager allocation

- Makes use of `handleSlotRequestFailedAtTaskManager` and simplifies it
because we can assume that a Slot is only registered once if we talk to
the same instance of a TaskExecutor. Further, we can omit to check the
free slots because we previously removed the slot from the free slots.

- `updateSlotStatus` only deals with new Task slots. All other updates
are performed by the `ResourceManager`. In case the ResourceManager
creashes, it will re-register all slot statuses.

commit c6768524c869ecdb84f0a8ae48837afc329c9714
Author: Maximilian Michels 
Date:   2016-09-29T14:03:39Z

[FLINK-4348] discard message from old TaskExecutorGateways

This fences old message using the InstanceID of the TaskExecutor.




> Implement slot allocation protocol with TaskExecutor
> 
>
> Key: FLINK-4348
> URL: https://issues.apache.org/jira/browse/FLINK-4348
> Project: Flink
>  Issue Type: Sub-task
>  Components: Cluster Management
>Reporter: Kurt Young
>Assignee: Maximilian Michels
>
> When slotManager finds a proper slot in the free pool for a slot request,  
> slotManager marks the slot as occupied, then tells the taskExecutor to give 
> the slot to the specified JobMaster. 
> when a slot request is sent to taskExecutor, it should contain following 
> parameters: AllocationID, JobID,  slotID, resourceManagerLeaderSessionID. 
> There exists 3 following possibilities of the response from taskExecutor, we 
> will discuss when each possibility happens and how to handle.
> 1. Ack request which means the taskExecutor gives the slot to the specified 
> jobMaster as expected.   
> 2. Decline request if the slot is already occupied by other AllocationID.  
> 3. Timeout which could caused by lost of request message or response message 
> or slow network transfer. 
> On the first occasion, ResourceManager need to do nothing. However, under the 
> second and third occasion, ResourceManager need to notify slotManager, 
> slotManager will verify and clear all the previous allocate information for 
> this slot request firstly, then try to find a proper slot for the slot 
> request again. This may cause some duplicate allocation, e.g. the slot 
> request to TaskManager is successful but the response is lost somehow, so we 
> may request a slot in another TaskManager, this causes two slots assigned to 
> one request, but it can be taken care of by rejecting registration at 
> JobMaster.
> There are still some question need to discuss in a step further.
> 1. Who send slotRequest to taskExecutor, SlotManager or ResourceManager? I 
> think it's better that SlotManager delegates the rpc call to ResourceManager 
> when SlotManager need to communicate with outside world.  ResourceManager 
> know which taskExecutor to send the request based on ResourceID. Besides this 
> RPC call which used to request slot to taskExecutor should not be a 
> RpcMethod,  because we hope only SlotManager has permission to call the 
> method, but the other component, for example JobMaster and TaskExecutor, 
> cannot call this method directly.
> 2. If 

[GitHub] flink pull request #2571: [FLINK-4348] Simplify logic of SlotManager

[mxm]

GitHub user mxm opened a pull request:

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

[FLINK-4348] Simplify logic of SlotManager

This pull request is split up into two commits:

1. It removes some code from the `SlotManager` to make it simpler. 
  - It makes use of `handleSlotRequestFailedAtTaskManager` and simplifies 
it because we can assume that a Slot is only registered once if we talk to the 
same instance of a TaskExecutor. Further, we can omit to check the free slots 
because we previously removed the slot from the free slots.
  - `updateSlotStatus` only deals with new Task slots. All other updates 
are performed by the `ResourceManager`. In case the ResourceManager creashes, 
it will re-register all slot statuses.

2. It fences `TaskExecutor` messages using an `InstanceID` which is 
required to make 1 work correctly. New messages have been introduced to achieve 
that.

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

$ git pull https://github.com/mxm/flink flip-6

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

https://github.com/apache/flink/pull/2571.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 #2571


commit 1d297403e5e085d944d80b498c4a269a794f8e2f
Author: Maximilian Michels 
Date:   2016-09-29T13:08:32Z

[FLINK-4347] change assumptions in SlotManager allocation

- Makes use of `handleSlotRequestFailedAtTaskManager` and simplifies it
because we can assume that a Slot is only registered once if we talk to
the same instance of a TaskExecutor. Further, we can omit to check the
free slots because we previously removed the slot from the free slots.

- `updateSlotStatus` only deals with new Task slots. All other updates
are performed by the `ResourceManager`. In case the ResourceManager
creashes, it will re-register all slot statuses.

commit c6768524c869ecdb84f0a8ae48837afc329c9714
Author: Maximilian Michels 
Date:   2016-09-29T14:03:39Z

[FLINK-4348] discard message from old TaskExecutorGateways

This fences old message using the InstanceID of the TaskExecutor.




---
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 #2569: [FLINK-4711] Let the Task trigger partition state ...

[uce]

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

https://github.com/apache/flink/pull/2569#discussion_r81164326
  
--- Diff: 
flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskControlMessages.scala
 ---
@@ -92,16 +92,6 @@ object TaskMessages {
   // 
--
 
   /**
-   * Answer to a [[RequestPartitionState]] with the state of the 
respective partition.
-   */
-  case class PartitionState(
--- End diff --

Has some unused imports after removal that we could remove before 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-4711) TaskManager can crash due to failing onPartitionStateUpdate call

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2569#discussion_r81162283
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/JobManagerCommunicationFactory.java
 ---
@@ -27,15 +26,6 @@
 public interface JobManagerCommunicationFactory {
--- End diff --

This has been become obsolete with your change and is unused. Let's remove 
it completely?


> TaskManager can crash due to failing onPartitionStateUpdate call
> 
>
> Key: FLINK-4711
> URL: https://issues.apache.org/jira/browse/FLINK-4711
> Project: Flink
>  Issue Type: Bug
>  Components: Distributed Coordination
>Affects Versions: 1.2.0
>Reporter: Till Rohrmann
>Assignee: Till Rohrmann
> Fix For: 1.2.0
>
>
> The {{TaskManager}} can crash because it calls 
> {{Task.onPartitionStateUpdate}} when it receives a {{PartitionState}} 
> message. The {{onPartitionStateUpdate}} method can throw an {{IOException}} 
> or {{InterruptedException}} which are not handled on the {{TaskManager}} 
> level.
> Another problem is that the initial partition state request is triggered 
> within the {{SingleInputGate}}. The request causes the {{JobManager}} to send 
> a {{PartitionState}} message to the {{TaskManager}} which forwards it to the 
> {{Task}}. If the at any of these points a message gets lost, then it is not 
> retried and the partition state remains unknown.
> In order to handle the exceptions, to make the data flow clearer and to add 
> automatic retries, I propose to let the {{Task}} send the partition state 
> check requests. Furthermore, the {{JobManager}} should directly answer to the 
> {{Task}} by replying to an ask operation. That way the message does not have 
> to be routed through the {{TaskManager}}.



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

[GitHub] flink pull request #2569: [FLINK-4711] Let the Task trigger partition state ...

[uce]

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

https://github.com/apache/flink/pull/2569#discussion_r81162283
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/JobManagerCommunicationFactory.java
 ---
@@ -27,15 +26,6 @@
 public interface JobManagerCommunicationFactory {
--- End diff --

This has been become obsolete with your change and is unused. Let's remove 
it completely?


---
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-4711) TaskManager can crash due to failing onPartitionStateUpdate call

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2569#discussion_r81164597
  
--- Diff: 
flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
 ---
@@ -873,8 +873,7 @@ class JobManager(
   }
 
   sender ! decorateMessage(
-PartitionState(
-  taskExecutionId,
+new org.apache.flink.runtime.io.network.PartitionState(
--- End diff --

I think it's not ambiguous anymore after removal of the TaskControlMessage 
and we can directly use `PartitionState`?


> TaskManager can crash due to failing onPartitionStateUpdate call
> 
>
> Key: FLINK-4711
> URL: https://issues.apache.org/jira/browse/FLINK-4711
> Project: Flink
>  Issue Type: Bug
>  Components: Distributed Coordination
>Affects Versions: 1.2.0
>Reporter: Till Rohrmann
>Assignee: Till Rohrmann
> Fix For: 1.2.0
>
>
> The {{TaskManager}} can crash because it calls 
> {{Task.onPartitionStateUpdate}} when it receives a {{PartitionState}} 
> message. The {{onPartitionStateUpdate}} method can throw an {{IOException}} 
> or {{InterruptedException}} which are not handled on the {{TaskManager}} 
> level.
> Another problem is that the initial partition state request is triggered 
> within the {{SingleInputGate}}. The request causes the {{JobManager}} to send 
> a {{PartitionState}} message to the {{TaskManager}} which forwards it to the 
> {{Task}}. If the at any of these points a message gets lost, then it is not 
> retried and the partition state remains unknown.
> In order to handle the exceptions, to make the data flow clearer and to add 
> automatic retries, I propose to let the {{Task}} send the partition state 
> check requests. Furthermore, the {{JobManager}} should directly answer to the 
> {{Task}} by replying to an ask operation. That way the message does not have 
> to be routed through the {{TaskManager}}.



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

[GitHub] flink pull request #2569: [FLINK-4711] Let the Task trigger partition state ...

[uce]

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

https://github.com/apache/flink/pull/2569#discussion_r81164597
  
--- Diff: 
flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
 ---
@@ -873,8 +873,7 @@ class JobManager(
   }
 
   sender ! decorateMessage(
-PartitionState(
-  taskExecutionId,
+new org.apache.flink.runtime.io.network.PartitionState(
--- End diff --

I think it's not ambiguous anymore after removal of the TaskControlMessage 
and we can directly use `PartitionState`?


---
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-4711) TaskManager can crash due to failing onPartitionStateUpdate call

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2569#discussion_r81164326
  
--- Diff: 
flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskControlMessages.scala
 ---
@@ -92,16 +92,6 @@ object TaskMessages {
   // 
--
 
   /**
-   * Answer to a [[RequestPartitionState]] with the state of the 
respective partition.
-   */
-  case class PartitionState(
--- End diff --

Has some unused imports after removal that we could remove before merging


> TaskManager can crash due to failing onPartitionStateUpdate call
> 
>
> Key: FLINK-4711
> URL: https://issues.apache.org/jira/browse/FLINK-4711
> Project: Flink
>  Issue Type: Bug
>  Components: Distributed Coordination
>Affects Versions: 1.2.0
>Reporter: Till Rohrmann
>Assignee: Till Rohrmann
> Fix For: 1.2.0
>
>
> The {{TaskManager}} can crash because it calls 
> {{Task.onPartitionStateUpdate}} when it receives a {{PartitionState}} 
> message. The {{onPartitionStateUpdate}} method can throw an {{IOException}} 
> or {{InterruptedException}} which are not handled on the {{TaskManager}} 
> level.
> Another problem is that the initial partition state request is triggered 
> within the {{SingleInputGate}}. The request causes the {{JobManager}} to send 
> a {{PartitionState}} message to the {{TaskManager}} which forwards it to the 
> {{Task}}. If the at any of these points a message gets lost, then it is not 
> retried and the partition state remains unknown.
> In order to handle the exceptions, to make the data flow clearer and to add 
> automatic retries, I propose to let the {{Task}} send the partition state 
> check requests. Furthermore, the {{JobManager}} should directly answer to the 
> {{Task}} by replying to an ask operation. That way the message does not have 
> to be routed through the {{TaskManager}}.



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

[jira] [Updated] (FLINK-4712) Implementing ranking predictions for ALS

[JIRA]

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

Domokos Miklós Kelen updated FLINK-4712:

Description: 
We started working on implementing ranking predictions for recommender systems. 
Ranking prediction means that beside predicting scores for user-item pairs, the 
recommender system is able to recommend a top K list for the users.

Details:

In practice, this would mean finding the K items for a particular user with the 
highest predicted rating. It should be possible also to specify whether to 
exclude the already seen items from a particular user's toplist. (See for 
example the 'exclude_known' setting of [Graphlab Create's ranking factorization 
recommender|https://turi.com/products/create/docs/generated/graphlab.recommender.ranking_factorization_recommender.RankingFactorizationRecommender.recommend.html#graphlab.recommender.ranking_factorization_recommender.RankingFactorizationRecommender.recommend]
 ).

The output of the topK recommendation function could be in the form of 
{{DataSet[(Int,Int,Int)]}}, meaning (user, item, rank), similar to Graphlab 
Create's output. However, this is arguable: follow up work includes 
implementing ranking recommendation evaluation metrics (such as precision@k, 
recall@k, ndcg@k), similar to [Spark's 
implementations|https://spark.apache.org/docs/1.5.0/mllib-evaluation-metrics.html#ranking-systems].
 It would be beneficial if we were able to design the API such that it could be 
included in the proposed evaluation framework (see 
[5157|https://issues.apache.org/jira/browse/FLINK-2157]), which makes it 
neccessary to consider the possible output type {{DataSet[(Int, Array[Int])]}} 
or {{DataSet[(Int, Array[(Int,Double)])]}} meaning (user, array of items), 
possibly including the predicted scores as well. See 
[4713|https://issues.apache.org/jira/browse/FLINK-4713] for details.

Another question arising is whether to provide this function as a member of the 
ALS class, as a switch-kind of parameter to the ALS implementation (meaning the 
model is either a rating or a ranking recommender model) or in some other way.

  was:
We started working on implementing ranking predictions for recommender systems. 
Ranking prediction means that beside predicting scores for user-item pairs, the 
recommender system is able to recommend a top K list for the users.

Details:

In practice, this would mean finding the K items for a particular user with the 
highest predicted rating. It should be possible also to specify whether to 
exclude the already seen items from a particular user's toplist. (See for 
example the 'exclude_known' setting of [Graphlab Create's ranking factorization 
recommender|https://turi.com/products/create/docs/generated/graphlab.recommender.ranking_factorization_recommender.RankingFactorizationRecommender.recommend.html#graphlab.recommender.ranking_factorization_recommender.RankingFactorizationRecommender.recommend]
 ).

The output of the topK recommendation function could be in the form of 
{{DataSet[(Int,Int,Int)]}}, meaning (user, item, rank), similar to Graphlab 
Create's output. However, this is arguable: follow up work includes 
implementing ranking recommendation evaluation metrics (such as precision@k, 
recall@k, ndcg@k), similar to [Spark's 
implementations|https://spark.apache.org/docs/1.5.0/mllib-evaluation-metrics.html#ranking-systems].
 It would be beneficial if we were able to design the API such that it could be 
included in the proposed evaluation framework (see 
[5157|https://issues.apache.org/jira/browse/FLINK-2157]), which makes it 
neccessary to consider the possible output type {{DataSet[(Int, Array[Int])]}} 
or {{DataSet[(Int, Array[(Int,Double)])]}} meaning (user, array of items), 
possibly including the predicted scores as well. See [issue todo] for details.

Another question arising is whether to provide this function as a member of the 
ALS class, as a switch-kind of parameter to the ALS implementation (meaning the 
model is either a rating or a ranking recommender model) or in some other way.


> Implementing ranking predictions for ALS
> 
>
> Key: FLINK-4712
> URL: https://issues.apache.org/jira/browse/FLINK-4712
> Project: Flink
>  Issue Type: New Feature
>  Components: Machine Learning Library
>Reporter: Domokos Miklós Kelen
>
> We started working on implementing ranking predictions for recommender 
> systems. Ranking prediction means that beside predicting scores for user-item 
> pairs, the recommender system is able to recommend a top K list for the users.
> Details:
> In practice, this would mean finding the K items for a particular user with 
> the highest predicted rating. It should be possible also to specify whether 
> to exclude the already seen items from a particular user's toplist. (See for 
> 

[GitHub] flink pull request #2570: [FLINK-3674] Add an interface for Time aware User ...

[aljoscha]

GitHub user aljoscha opened a pull request:

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

[FLINK-3674] Add an interface for Time aware User Functions

This moves the event-time/processing-time trigger code from
`WindowOperator` behind a well defined interface that can be used by
operators (and user functions).

`InternalTimerService` is the new interface that has the same
functionality that `WindowOperator` used to have. `TimerService` is the user
facing interface that does not allow dealing with namespaces/payloads
and also does not allow deleting timers. There is a default
implementation in `HeapInternalTimerService` that can checkpoint timers to
a stream and also restore from a stream. Right now, this is managed in
`AbstractStreamOperator` and operators can ask for an
`InternalTimerService`.

This also adds tests for `HeapInternalTimerService`.

This adds two new user functions:
 - `TimelyFlatMapFunction`: an extension of `FlatMapFunction` that also
   allows querying time and setting timers
 - `TimelyCoFlatMapFunction`: the same, but for `CoFlatMapFunction`

There are two new `StreamOperator` implementations for these that use the
`InternalTimerService` interface.

This also adds tests for the two new operators.

This also adds the new interface `KeyContext` that is used for
setting/querying the current key context for state and timers. Timers
are always scoped to a key, for now.

Also, this moves the handling of watermarks for both one-input and
two-input operators to `AbstractStreamOperators` so that we have a central
ground-truth.

There was also a bunch of small changes that I had to do to make the proper 
change more clean. I would like to keep these as separate commits because they 
clearly document what was going on.

## Note for Reviewers
You should probably start from the tests, i.e. 
`HeapInternalTimerServiceTest`, `TimelyFlatMapTest` and `TimelyCoFlatMapTest`. 
Then, the other interesting bits are `AbstractStreamOperator` that now deals 
with watermarks and checkpointing the timers and the `HeapInternalTimerService` 
as well. Keep in mind that this is just moving the code from `WindowOperator` 
to `HeapInternalTimerService` with some generalizations. I didn't try to 
optimize any of the data structures that are used.

R: @StephanEwen @StefanRRichter @kl0u for review, please 😃 

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

$ git pull https://github.com/aljoscha/flink timely-function

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

https://github.com/apache/flink/pull/2570.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 #2570


commit 1a09d9032bf5683a378a7fc8dc480f2d14c5924d
Author: Aljoscha Krettek 
Date:   2016-09-25T18:58:16Z

Rename TimeServiceProvider to ProcessingTimeService

The name is clashing with the soon-to-be-added
TimerService/InternalTimerService which is meant as an interface for
dealing with both processing time and event time.

TimeServiceProvided is renamed to ProcessingTimeService to reflect the
fact that it is a low-level utility that only deals with "physical"
processing-time trigger tasks.

commit 758827c3c8508ef9ef2ec079ff3a8469d0096ca8
Author: Aljoscha Krettek 
Date:   2016-09-28T13:10:35Z

Use OperatorTestHarness and TestProcessingTimeService in Kafka Tests

commit f6dd9c74dc2c58c4263fb6d084651b514898d47a
Author: Aljoscha Krettek 
Date:   2016-09-28T14:35:33Z

Use Processing-Time Service of TestHarness in WindowOperatorTest

Before, this was manually creating a TestProcessingTimeService, now,
we're using the one that is there by default in
OneInputStreamOperatorTestHarness.

commit 65389d66c5586e6707b7a6bf48df512354fac085
Author: Aljoscha Krettek 
Date:   2016-09-28T14:43:40Z

Refactor OperatorTestHarness to always use TestProcessingTimeService

Before, this would allow handing in a custom ProcessingTimeService but
this was in reality always TestProcessingTimeService.

commit 1d013bcacc040552e5783c64d094ec309014457b
Author: Aljoscha Krettek 
Date:   2016-09-28T13:12:26Z

Use TestHarness Processing-time Facility in BucketingSinkTest

Before, this was manually creating a TestProcessingTimeService. Now we
use the one that is there by default in
OneInputStreamOperatorTestHarness.

commit eaf3dd00fefeb2487c7cafff6337123cbe42874b
Author: Aljoscha Krettek 
Date:   2016-09-28T13:32:24Z

Use OperatorTestHarness in AlignedWindowOperator Tests

commit b597d2ef50c27554b83fddaff8873107265340d4

[jira] [Commented] (FLINK-4068) Move constant computations out of code-generated `flatMap` functions.

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2560#discussion_r81163353
  
--- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkPlannerImpl.scala
 ---
@@ -60,6 +60,7 @@ class FlinkPlannerImpl(
   var root: RelRoot = _
 
   private def ready() {
+planner.setExecutor(config.getExecutor)
--- End diff --

I would not set the planner here as this class is only used by the SQL API. 
`FlinkRelBuilder.create()` is a better place.


> Move constant computations out of code-generated `flatMap` functions.
> -
>
> Key: FLINK-4068
> URL: https://issues.apache.org/jira/browse/FLINK-4068
> Project: Flink
>  Issue Type: Improvement
>  Components: Table API & SQL
>Affects Versions: 1.1.0
>Reporter: Fabian Hueske
>Assignee: Jark Wu
>
> The generated functions for expressions of the Table API or SQL include 
> constant computations.
> For instance the code generated for a predicate like:
> {code}
> myInt < (10 + 20)
> {code}
> looks roughly like:
> {code}
> public void flatMap(Row in, Collector out) {
>   Integer in1 = in.productElement(1);
>   int temp = 10 + 20;  
>   if (in1 < temp) {
> out.collect(in)
>   }
> }
> {code}
> In this example the computation of {{temp}} is constant and could be moved 
> out of the {{flatMap()}} method.
> The same might apply for generated function other than {{FlatMap}} as well.



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

[GitHub] flink pull request #2560: [FLINK-4068] [table] Move constant computations ou...

[twalthr]

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

https://github.com/apache/flink/pull/2560#discussion_r81163353
  
--- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkPlannerImpl.scala
 ---
@@ -60,6 +60,7 @@ class FlinkPlannerImpl(
   var root: RelRoot = _
 
   private def ready() {
+planner.setExecutor(config.getExecutor)
--- End diff --

I would not set the planner here as this class is only used by the SQL API. 
`FlinkRelBuilder.create()` is a better place.


---
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-4712) Implementing ranking predictions for ALS

[JIRA]

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

Domokos Miklós Kelen updated FLINK-4712:

Description: 
We started working on implementing ranking predictions for recommender systems. 
Ranking prediction means that beside predicting scores for user-item pairs, the 
recommender system is able to recommend a top K list for the users.

Details:

In practice, this would mean finding the K items for a particular user with the 
highest predicted rating. It should be possible also to specify whether to 
exclude the already seen items from a particular user's toplist. (See for 
example the 'exclude_known' setting of [Graphlab Create's ranking factorization 
recommender|https://turi.com/products/create/docs/generated/graphlab.recommender.ranking_factorization_recommender.RankingFactorizationRecommender.recommend.html#graphlab.recommender.ranking_factorization_recommender.RankingFactorizationRecommender.recommend].

The output of the topK recommendation function could be in the form of 
{{DataSet[(Int,Int,Int)]}}, meaning (user, item, rank), similar to Graphlab 
Create's output. However, this is arguable: follow up work includes 
implementing ranking recommendation evaluation metrics (such as precision@k, 
recall@k, ndcg@k), similar to [Spark's 
implementations|https://spark.apache.org/docs/1.5.0/mllib-evaluation-metrics.html#ranking-systems].
 It would be beneficial if we were able to design the API such that it could be 
included in the proposed evaluation framework (see 
[5157|https://issues.apache.org/jira/browse/FLINK-2157]), which makes it 
neccessary to consider the possible output type {{DataSet[(Int, Array[Int])]}} 
or {{DataSet[(Int, Array[(Int,Double)])]}} meaning (user, array of items), 
possibly including the predicted scores as well. See [issue todo] for details.

Another question arising is whether to provide this function as a member of the 
ALS class, as a switch-kind of parameter to the ALS implementation (meaning the 
model is either a rating or a ranking recommender model) or in some other way.

  was:
We started working on implementing ranking predictions for recommender systems. 
Ranking prediction means that beside predicting scores for user-item pairs, the 
recommender system is able to recommend a top K list for the users.

Details:

In practice, this would mean finding the K items for a particular user with the 
highest predicted rating. It should be possible also to specify whether to 
exclude the already seen items from a particular user's toplist. (See for 
example the 'exclude_known' setting of [Graphlab Create's ranking factorization 
recommender|https://turi.com/products/create/docs/generated/graphlab.recommender.ranking_factorization_recommender.RankingFactorizationRecommender.recommend.html#graphlab.recommender.ranking_factorization_recommender.RankingFactorizationRecommender.recommend].

The output of the topK recommendation function could be in the form of 
DataSet[(Int,Int,Int)], meaning (user, item, rank), similar to Graphlab 
Create's output. However, this is arguable: follow up work includes 
implementing ranking recommendation evaluation metrics (such as precision@k, 
recall@k, ndcg@k), similar to [Spark's 
implementations|https://spark.apache.org/docs/1.5.0/mllib-evaluation-metrics.html#ranking-systems].
 It would be beneficial if we were able to design the API such that it could be 
included in the proposed evaluation framework (see 
[5157|https://issues.apache.org/jira/browse/FLINK-2157]), which makes it 
neccessary to consider the possible output type DataSet[(Int, Array[Int])] or 
DataSet[(Int, Array[(Int,Double)])] meaning (user, array of items), possibly 
including the predicted scores as well. See [issue todo] for details.

Another question arising is whether to provide this function as a member of the 
ALS class, as a switch-kind of parameter to the ALS implementation (meaning the 
model is either a rating or a ranking recommender model) or in some other way.


> Implementing ranking predictions for ALS
> 
>
> Key: FLINK-4712
> URL: https://issues.apache.org/jira/browse/FLINK-4712
> Project: Flink
>  Issue Type: New Feature
>  Components: Machine Learning Library
>Reporter: Domokos Miklós Kelen
>
> We started working on implementing ranking predictions for recommender 
> systems. Ranking prediction means that beside predicting scores for user-item 
> pairs, the recommender system is able to recommend a top K list for the users.
> Details:
> In practice, this would mean finding the K items for a particular user with 
> the highest predicted rating. It should be possible also to specify whether 
> to exclude the already seen items from a particular user's toplist. (See for 
> example the 'exclude_known' setting of [Graphlab Create's 

[jira] [Commented] (FLINK-4068) Move constant computations out of code-generated `flatMap` functions.

[ASF GitHub Bot (JIRA)]

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

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

Github user twalthr commented on the issue:

https://github.com/apache/flink/pull/2560
  
@wuchong it seems that the constant expression reduction does not work in 
the Table API . Could you convert your SQL tests also to additional Table tests 
and have a look at it? The SQL part seems to work fine.


> Move constant computations out of code-generated `flatMap` functions.
> -
>
> Key: FLINK-4068
> URL: https://issues.apache.org/jira/browse/FLINK-4068
> Project: Flink
>  Issue Type: Improvement
>  Components: Table API & SQL
>Affects Versions: 1.1.0
>Reporter: Fabian Hueske
>Assignee: Jark Wu
>
> The generated functions for expressions of the Table API or SQL include 
> constant computations.
> For instance the code generated for a predicate like:
> {code}
> myInt < (10 + 20)
> {code}
> looks roughly like:
> {code}
> public void flatMap(Row in, Collector out) {
>   Integer in1 = in.productElement(1);
>   int temp = 10 + 20;  
>   if (in1 < temp) {
> out.collect(in)
>   }
> }
> {code}
> In this example the computation of {{temp}} is constant and could be moved 
> out of the {{flatMap()}} method.
> The same might apply for generated function other than {{FlatMap}} as well.



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

[jira] [Created] (FLINK-4713) Implementing ranking evaluation scores for recommender systems

[JIRA]

Domokos Miklós Kelen created FLINK-4713:
---

 Summary: Implementing ranking evaluation scores for recommender 
systems
 Key: FLINK-4713
 URL: https://issues.apache.org/jira/browse/FLINK-4713
 Project: Flink
  Issue Type: New Feature
  Components: Machine Learning Library
Reporter: Domokos Miklós Kelen


Follow up work to [4712|https://issues.apache.org/jira/browse/FLINK-4712] 
includes implementing ranking recommendation evaluation metrics (such as 
precision@k, recall@k, ndcg@k), [similar to Spark's 
implementations|https://spark.apache.org/docs/1.5.0/mllib-evaluation-metrics.html#ranking-systems].
 It would be beneficial if we were able to design the API such that it could be 
included in the proposed evaluation framework (see 
[2157|https://issues.apache.org/jira/browse/FLINK-2157]).

In it's current form, this would mean generalizing the PredictionType type 
parameter of the Score class to allow for {{Array[Int]}} or {{Array[(Int, 
Double)]}}, and outputting the recommendations in the form {{DataSet[(Int, 
Array[Int])]}} or {{DataSet[(Int, Array[(Int,Double)])]}} meaning (user, array 
of items), possibly including the predicted scores as well. 

However, calculating for example nDCG for a given user u requires us to be able 
to access all of the (u, item, relevance) records in the test dataset, which 
means we would need to put this information in the second element of the 
{{DataSet[(PredictionType, PredictionType)]}} input of the scorer function as 
PredictionType={{Array[(Int, Double)]}}. This is problematic, as this Array 
could be arbitrarily long.

Another option is to further rework the proposed evaluation framework to allow 
us to implement this properly, with inputs in the form of {{recommendations : 
DataSet[(Int,Int,Int)]}} (user, item, rank) and {{test : 
DataSet[(Int,Int,Double)]}} (user, item relevance). This way, the scores could 
be implemented such that they can be calculated in a distributed way.

The third option is to implement the scorer functions outside the evaluation 
framework.



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

[GitHub] flink issue #2542: [FLINK-4613] [ml] Extend ALS to handle implicit feedback ...

[thvasilo]

Github user thvasilo commented on the issue:

https://github.com/apache/flink/pull/2542
  
@gaborhermann I think have a larger scale test would boost our confidence 
in the implementation, and maybe point out some problems that do not manifest 
with small data, which is very common.

If you plan to do it anyway we might as well do it before merging the code, 
but it's not a blocker.


---
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-4712) Implementing ranking predictions for ALS

[JIRA]

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

Domokos Miklós Kelen updated FLINK-4712:

Description: 
We started working on implementing ranking predictions for recommender systems. 
Ranking prediction means that beside predicting scores for user-item pairs, the 
recommender system is able to recommend a top K list for the users.

Details:

In practice, this would mean finding the K items for a particular user with the 
highest predicted rating. It should be possible also to specify whether to 
exclude the already seen items from a particular user's toplist. (See for 
example the 'exclude_known' setting of [Graphlab Create's ranking factorization 
recommender|https://turi.com/products/create/docs/generated/graphlab.recommender.ranking_factorization_recommender.RankingFactorizationRecommender.recommend.html#graphlab.recommender.ranking_factorization_recommender.RankingFactorizationRecommender.recommend]
 ).

The output of the topK recommendation function could be in the form of 
{{DataSet[(Int,Int,Int)]}}, meaning (user, item, rank), similar to Graphlab 
Create's output. However, this is arguable: follow up work includes 
implementing ranking recommendation evaluation metrics (such as precision@k, 
recall@k, ndcg@k), similar to [Spark's 
implementations|https://spark.apache.org/docs/1.5.0/mllib-evaluation-metrics.html#ranking-systems].
 It would be beneficial if we were able to design the API such that it could be 
included in the proposed evaluation framework (see 
[5157|https://issues.apache.org/jira/browse/FLINK-2157]), which makes it 
neccessary to consider the possible output type {{DataSet[(Int, Array[Int])]}} 
or {{DataSet[(Int, Array[(Int,Double)])]}} meaning (user, array of items), 
possibly including the predicted scores as well. See [issue todo] for details.

Another question arising is whether to provide this function as a member of the 
ALS class, as a switch-kind of parameter to the ALS implementation (meaning the 
model is either a rating or a ranking recommender model) or in some other way.

  was:
We started working on implementing ranking predictions for recommender systems. 
Ranking prediction means that beside predicting scores for user-item pairs, the 
recommender system is able to recommend a top K list for the users.

Details:

In practice, this would mean finding the K items for a particular user with the 
highest predicted rating. It should be possible also to specify whether to 
exclude the already seen items from a particular user's toplist. (See for 
example the 'exclude_known' setting of [Graphlab Create's ranking factorization 
recommender|https://turi.com/products/create/docs/generated/graphlab.recommender.ranking_factorization_recommender.RankingFactorizationRecommender.recommend.html#graphlab.recommender.ranking_factorization_recommender.RankingFactorizationRecommender.recommend].

The output of the topK recommendation function could be in the form of 
{{DataSet[(Int,Int,Int)]}}, meaning (user, item, rank), similar to Graphlab 
Create's output. However, this is arguable: follow up work includes 
implementing ranking recommendation evaluation metrics (such as precision@k, 
recall@k, ndcg@k), similar to [Spark's 
implementations|https://spark.apache.org/docs/1.5.0/mllib-evaluation-metrics.html#ranking-systems].
 It would be beneficial if we were able to design the API such that it could be 
included in the proposed evaluation framework (see 
[5157|https://issues.apache.org/jira/browse/FLINK-2157]), which makes it 
neccessary to consider the possible output type {{DataSet[(Int, Array[Int])]}} 
or {{DataSet[(Int, Array[(Int,Double)])]}} meaning (user, array of items), 
possibly including the predicted scores as well. See [issue todo] for details.

Another question arising is whether to provide this function as a member of the 
ALS class, as a switch-kind of parameter to the ALS implementation (meaning the 
model is either a rating or a ranking recommender model) or in some other way.


> Implementing ranking predictions for ALS
> 
>
> Key: FLINK-4712
> URL: https://issues.apache.org/jira/browse/FLINK-4712
> Project: Flink
>  Issue Type: New Feature
>  Components: Machine Learning Library
>Reporter: Domokos Miklós Kelen
>
> We started working on implementing ranking predictions for recommender 
> systems. Ranking prediction means that beside predicting scores for user-item 
> pairs, the recommender system is able to recommend a top K list for the users.
> Details:
> In practice, this would mean finding the K items for a particular user with 
> the highest predicted rating. It should be possible also to specify whether 
> to exclude the already seen items from a particular user's toplist. (See for 
> example the 'exclude_known' setting of 

[jira] [Created] (FLINK-4712) Implementing ranking predictions for ALS

[JIRA]

Domokos Miklós Kelen created FLINK-4712:
---

 Summary: Implementing ranking predictions for ALS
 Key: FLINK-4712
 URL: https://issues.apache.org/jira/browse/FLINK-4712
 Project: Flink
  Issue Type: New Feature
  Components: Machine Learning Library
Reporter: Domokos Miklós Kelen


We started working on implementing ranking predictions for recommender systems. 
Ranking prediction means that beside predicting scores for user-item pairs, the 
recommender system is able to recommend a top K list for the users.

Details:

In practice, this would mean finding the K items for a particular user with the 
highest predicted rating. It should be possible also to specify whether to 
exclude the already seen items from a particular user's toplist. (See for 
example the 'exclude_known' setting of [Graphlab Create's ranking factorization 
recommender|https://turi.com/products/create/docs/generated/graphlab.recommender.ranking_factorization_recommender.RankingFactorizationRecommender.recommend.html#graphlab.recommender.ranking_factorization_recommender.RankingFactorizationRecommender.recommend].

The output of the topK recommendation function could be in the form of 
DataSet[(Int,Int,Int)], meaning (user, item, rank), similar to Graphlab 
Create's output. However, this is arguable: follow up work includes 
implementing ranking recommendation evaluation metrics (such as precision@k, 
recall@k, ndcg@k), similar to [Spark's 
implementations|https://spark.apache.org/docs/1.5.0/mllib-evaluation-metrics.html#ranking-systems].
 It would be beneficial if we were able to design the API such that it could be 
included in the proposed evaluation framework (see 
[5157|https://issues.apache.org/jira/browse/FLINK-2157]), which makes it 
neccessary to consider the possible output type DataSet[(Int, Array[Int])] or 
DataSet[(Int, Array[(Int,Double)])] meaning (user, array of items), possibly 
including the predicted scores as well. See [issue todo] for details.

Another question arising is whether to provide this function as a member of the 
ALS class, as a switch-kind of parameter to the ALS implementation (meaning the 
model is either a rating or a ranking recommender model) or in some other way.



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

[jira] [Commented] (FLINK-4613) Extend ALS to handle implicit feedback datasets

[ASF GitHub Bot (JIRA)]

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

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

Github user thvasilo commented on the issue:

https://github.com/apache/flink/pull/2542
  
@gaborhermann I think have a larger scale test would boost our confidence 
in the implementation, and maybe point out some problems that do not manifest 
with small data, which is very common.

If you plan to do it anyway we might as well do it before merging the code, 
but it's not a blocker.


> Extend ALS to handle implicit feedback datasets
> ---
>
> Key: FLINK-4613
> URL: https://issues.apache.org/jira/browse/FLINK-4613
> Project: Flink
>  Issue Type: New Feature
>  Components: Machine Learning Library
>Reporter: Gábor Hermann
>Assignee: Gábor Hermann
>
> The Alternating Least Squares implementation should be extended to handle 
> _implicit feedback_ datasets. These datasets do not contain explicit ratings 
> by users, they are rather built by collecting user behavior (e.g. user 
> listened to artist X for Y minutes), and they require a slightly different 
> optimization objective. See details by [Hu et 
> al|http://dx.doi.org/10.1109/ICDM.2008.22].
> We do not need to modify much in the original ALS algorithm. See [Spark ALS 
> implementation|https://github.com/apache/spark/blob/master/mllib/src/main/scala/org/apache/spark/ml/recommendation/ALS.scala],
>  which could be a basis for this extension. Only the updating factor part is 
> modified, and most of the changes are in the local parts of the algorithm 
> (i.e. UDFs). In fact, the only modification that is not local, is 
> precomputing a matrix product Y^T * Y and broadcasting it to all the nodes, 
> which we can do with broadcast DataSets. 



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

[GitHub] flink pull request #2547: [FLINK-4552] Refactor WindowOperator/Trigger Tests

[aljoscha]

Github user aljoscha closed the pull request at:

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


---
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-4613) Extend ALS to handle implicit feedback datasets

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2542#discussion_r81159482
  
--- Diff: 
flink-libraries/flink-ml/src/test/scala/org/apache/flink/ml/recommendation/ImplicitALSTest.scala
 ---
@@ -0,0 +1,171 @@
+/*
+ * 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.ml.recommendation
+
+import org.apache.flink.ml.util.FlinkTestBase
+import org.scalatest._
+
+import scala.language.postfixOps
+import org.apache.flink.api.scala._
+import org.apache.flink.core.testutils.CommonTestUtils
+
+class ImplicitALSTest
+  extends FlatSpec
+with Matchers
+with FlinkTestBase {
+
+  override val parallelism = 2
+
+  behavior of "The modification of the alternating least squares (ALS) 
implementation" +
+"for implicit feedback datasets."
+
+  it should "properly compute Y^T * Y, and factorize matrix" in {
--- End diff --

AFAIK in the rest of the FlinkML tests we just use `val env = 
ExecutionEnvironment.getExecutionEnvironment`. I don't know if that policy has 
now changed, maybe @tillrohrmann can clarify.

For now I would say to just split the tests.


> Extend ALS to handle implicit feedback datasets
> ---
>
> Key: FLINK-4613
> URL: https://issues.apache.org/jira/browse/FLINK-4613
> Project: Flink
>  Issue Type: New Feature
>  Components: Machine Learning Library
>Reporter: Gábor Hermann
>Assignee: Gábor Hermann
>
> The Alternating Least Squares implementation should be extended to handle 
> _implicit feedback_ datasets. These datasets do not contain explicit ratings 
> by users, they are rather built by collecting user behavior (e.g. user 
> listened to artist X for Y minutes), and they require a slightly different 
> optimization objective. See details by [Hu et 
> al|http://dx.doi.org/10.1109/ICDM.2008.22].
> We do not need to modify much in the original ALS algorithm. See [Spark ALS 
> implementation|https://github.com/apache/spark/blob/master/mllib/src/main/scala/org/apache/spark/ml/recommendation/ALS.scala],
>  which could be a basis for this extension. Only the updating factor part is 
> modified, and most of the changes are in the local parts of the algorithm 
> (i.e. UDFs). In fact, the only modification that is not local, is 
> precomputing a matrix product Y^T * Y and broadcasting it to all the nodes, 
> which we can do with broadcast DataSets. 



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

[GitHub] flink pull request #2542: [FLINK-4613] [ml] Extend ALS to handle implicit fe...

[thvasilo]

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

https://github.com/apache/flink/pull/2542#discussion_r81159482
  
--- Diff: 
flink-libraries/flink-ml/src/test/scala/org/apache/flink/ml/recommendation/ImplicitALSTest.scala
 ---
@@ -0,0 +1,171 @@
+/*
+ * 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.ml.recommendation
+
+import org.apache.flink.ml.util.FlinkTestBase
+import org.scalatest._
+
+import scala.language.postfixOps
+import org.apache.flink.api.scala._
+import org.apache.flink.core.testutils.CommonTestUtils
+
+class ImplicitALSTest
+  extends FlatSpec
+with Matchers
+with FlinkTestBase {
+
+  override val parallelism = 2
+
+  behavior of "The modification of the alternating least squares (ALS) 
implementation" +
+"for implicit feedback datasets."
+
+  it should "properly compute Y^T * Y, and factorize matrix" in {
--- End diff --

AFAIK in the rest of the FlinkML tests we just use `val env = 
ExecutionEnvironment.getExecutionEnvironment`. I don't know if that policy has 
now changed, maybe @tillrohrmann can clarify.

For now I would say to just split the tests.


---
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-4613) Extend ALS to handle implicit feedback datasets

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/2542#discussion_r81159171
  
--- Diff: 
flink-libraries/flink-ml/src/main/scala/org/apache/flink/ml/recommendation/ALS.scala
 ---
@@ -581,6 +637,16 @@ object ALS {
 val userXy = new ArrayBuffer[Array[Double]]()
 val numRatings = new ArrayBuffer[Int]()
 
+var precomputedXtX: Array[Double] = null
+
+override def open(config: Configuration): Unit = {
+  // retrieve broadcasted precomputed XtX if using implicit 
feedback
+  if (implicitPrefs) {
+precomputedXtX = 
getRuntimeContext.getBroadcastVariable[Array[Double]]("XtX")
+  .iterator().next()
+  }
+}
+
 override def coGroup(left: lang.Iterable[(Int, Int, 
Array[Array[Double]])],
--- End diff --

We can ping @tillrohrmann here, as the original author maybe he has some 
input.


> Extend ALS to handle implicit feedback datasets
> ---
>
> Key: FLINK-4613
> URL: https://issues.apache.org/jira/browse/FLINK-4613
> Project: Flink
>  Issue Type: New Feature
>  Components: Machine Learning Library
>Reporter: Gábor Hermann
>Assignee: Gábor Hermann
>
> The Alternating Least Squares implementation should be extended to handle 
> _implicit feedback_ datasets. These datasets do not contain explicit ratings 
> by users, they are rather built by collecting user behavior (e.g. user 
> listened to artist X for Y minutes), and they require a slightly different 
> optimization objective. See details by [Hu et 
> al|http://dx.doi.org/10.1109/ICDM.2008.22].
> We do not need to modify much in the original ALS algorithm. See [Spark ALS 
> implementation|https://github.com/apache/spark/blob/master/mllib/src/main/scala/org/apache/spark/ml/recommendation/ALS.scala],
>  which could be a basis for this extension. Only the updating factor part is 
> modified, and most of the changes are in the local parts of the algorithm 
> (i.e. UDFs). In fact, the only modification that is not local, is 
> precomputing a matrix product Y^T * Y and broadcasting it to all the nodes, 
> which we can do with broadcast DataSets. 



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

[GitHub] flink pull request #2542: [FLINK-4613] [ml] Extend ALS to handle implicit fe...

[thvasilo]

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

https://github.com/apache/flink/pull/2542#discussion_r81159171
  
--- Diff: 
flink-libraries/flink-ml/src/main/scala/org/apache/flink/ml/recommendation/ALS.scala
 ---
@@ -581,6 +637,16 @@ object ALS {
 val userXy = new ArrayBuffer[Array[Double]]()
 val numRatings = new ArrayBuffer[Int]()
 
+var precomputedXtX: Array[Double] = null
+
+override def open(config: Configuration): Unit = {
+  // retrieve broadcasted precomputed XtX if using implicit 
feedback
+  if (implicitPrefs) {
+precomputedXtX = 
getRuntimeContext.getBroadcastVariable[Array[Double]]("XtX")
+  .iterator().next()
+  }
+}
+
 override def coGroup(left: lang.Iterable[(Int, Int, 
Array[Array[Double]])],
--- End diff --

We can ping @tillrohrmann here, as the original author maybe he has some 
input.


---
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.
---