[jira] [Commented] (FLINK-2186) Rework CSV import to support very wide files

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/3012#discussion_r100247899
  
--- Diff: 
flink-tests/src/test/java/org/apache/flink/test/io/CsvReaderITCase.java ---
@@ -122,6 +127,80 @@ public void testValueTypes() throws Exception {
compareResultAsTuples(result, expected);
}
 
+   private int fullRowSize = 29;
+   private String including = "011";
+   private String fileContent =
+   "1,2,3," + 4 + "," + 5.0d + "," + true +
+   ",7,8,9,11,22,33,44,55,66,77,88,99,00," +
+   "111,222,333,444,555,666,777,888,999,000\n" +
+   "a,b,c," + 40 + "," + 50.0d + "," + false +
+   ",g,h,i,aa,bb,cc,dd,ee,ff,gg,hh,ii,mm," +
+   "aaa,bbb,ccc,ddd,eee,fff,ggg,hhh,iii,mmm\n";
--- End diff --

A lot of concatenations, the fileContent equals 
```
"1,2,3,4,5.0,true" +
",7,8,9,11,22,33,44,55,66,77,88,99,00," +
"111,222,333,444,555,666,777,888,999,000\n" +
"a,b,c,40,50.0,false," +
"g,h,i,aa,bb,cc,dd,ee,ff,gg,hh,ii,mm," +
"aaa,bbb,ccc,ddd,eee,fff,ggg,hhh,iii,mmm\n"
```


> Rework CSV import to support very wide files
> 
>
> Key: FLINK-2186
> URL: https://issues.apache.org/jira/browse/FLINK-2186
> Project: Flink
>  Issue Type: Improvement
>  Components: Machine Learning Library, Scala API
>Reporter: Theodore Vasiloudis
>Assignee: Anton Solovev
>
> In the current readVcsFile implementation, importing CSV files with many 
> columns can become from cumbersome to impossible.
> For example to import an 11 column file we need to write:
> {code}
> val cancer = env.readCsvFile[(String, String, String, String, String, String, 
> String, String, String, String, 
> String)]("/path/to/breast-cancer-wisconsin.data")
> {code}
> For many use cases in Machine Learning we might have CSV files with thousands 
> or millions of columns that we want to import as vectors.
> In that case using the current readCsvFile method becomes impossible.
> We therefore need to rework the current function, or create a new one that 
> will allow us to import CSV files with an arbitrary number of columns.



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

[jira] [Commented] (FLINK-2186) Rework CSV import to support very wide files

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/3012#discussion_r100247879
  
--- Diff: 
flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
 ---
@@ -348,6 +349,47 @@ class ExecutionEnvironment(javaEnv: JavaEnv) {
 wrap(new DataSource[T](javaEnv, inputFormat, typeInfo, 
getCallLocationName()))
   }
 
+  def readCsvFileAsRow[T : ClassTag : TypeInformation](
+filePath: String,
+rowSize: Int,
+additionalTypes: Map[Int, Class[_]] = null,
+lineDelimiter: String = "\n",
+fieldDelimiter: String = ",",
+quoteCharacter: Character = null,
+ignoreFirstLine: Boolean = false,
+ignoreComments: String = null,
+lenient: Boolean = false,
+includedFields: Array[Int] = null): DataSet[Row] = {
--- End diff --

Please add more two spase for parameters like as for other methods in 
class, space chars should is 6.


> Rework CSV import to support very wide files
> 
>
> Key: FLINK-2186
> URL: https://issues.apache.org/jira/browse/FLINK-2186
> Project: Flink
>  Issue Type: Improvement
>  Components: Machine Learning Library, Scala API
>Reporter: Theodore Vasiloudis
>Assignee: Anton Solovev
>
> In the current readVcsFile implementation, importing CSV files with many 
> columns can become from cumbersome to impossible.
> For example to import an 11 column file we need to write:
> {code}
> val cancer = env.readCsvFile[(String, String, String, String, String, String, 
> String, String, String, String, 
> String)]("/path/to/breast-cancer-wisconsin.data")
> {code}
> For many use cases in Machine Learning we might have CSV files with thousands 
> or millions of columns that we want to import as vectors.
> In that case using the current readCsvFile method becomes impossible.
> We therefore need to rework the current function, or create a new one that 
> will allow us to import CSV files with an arbitrary number of columns.



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

[jira] [Commented] (FLINK-2186) Rework CSV import to support very wide files

[ASF GitHub Bot (JIRA)]

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

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

Github user ex00 commented on the issue:

https://github.com/apache/flink/pull/3012
  
Thanks for work  @tonycox.
The PR looks good to me. I left a few minor comments about PR.

What do you mean about test negative case? If typeMap does not match with 
fields type in file for example


> Rework CSV import to support very wide files
> 
>
> Key: FLINK-2186
> URL: https://issues.apache.org/jira/browse/FLINK-2186
> Project: Flink
>  Issue Type: Improvement
>  Components: Machine Learning Library, Scala API
>Reporter: Theodore Vasiloudis
>Assignee: Anton Solovev
>
> In the current readVcsFile implementation, importing CSV files with many 
> columns can become from cumbersome to impossible.
> For example to import an 11 column file we need to write:
> {code}
> val cancer = env.readCsvFile[(String, String, String, String, String, String, 
> String, String, String, String, 
> String)]("/path/to/breast-cancer-wisconsin.data")
> {code}
> For many use cases in Machine Learning we might have CSV files with thousands 
> or millions of columns that we want to import as vectors.
> In that case using the current readCsvFile method becomes impossible.
> We therefore need to rework the current function, or create a new one that 
> will allow us to import CSV files with an arbitrary number of columns.



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

[GitHub] flink pull request #3012: [FLINK-2186] Add readCsvAsRow methods to CsvReader...

[ex00]

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

https://github.com/apache/flink/pull/3012#discussion_r100247845
  
--- Diff: 
flink-java/src/main/java/org/apache/flink/api/java/io/CsvReader.java ---
@@ -351,6 +356,32 @@ public CsvReader ignoreInvalidLines(){
return new DataSource(executionContext, inputFormat, 
typeInfo, Utils.getCallLocationName());
}

+   public DataSource rowType(Class mainTargetType, int size, 
Map additionalTypes) {
--- End diff --

Please add javadoc


---
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 #3012: [FLINK-2186] Add readCsvAsRow methods to CsvReader...

[ex00]

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

https://github.com/apache/flink/pull/3012#discussion_r100247856
  
--- Diff: 
flink-java/src/main/java/org/apache/flink/api/java/io/CsvReader.java ---
@@ -351,6 +356,32 @@ public CsvReader ignoreInvalidLines(){
return new DataSource(executionContext, inputFormat, 
typeInfo, Utils.getCallLocationName());
}

+   public DataSource rowType(Class mainTargetType, int size, 
Map additionalTypes) {
+   Preconditions.checkNotNull(mainTargetType, "The main type class 
must not be null.");
+
+   TypeInformation typeInfo = 
TypeExtractor.createTypeInfo(mainTargetType);
+   RowTypeInfo rowTypeInfo;
+
+   if (additionalTypes != null) {
+   Map addTypeMap = new 
HashMap<>(additionalTypes.size());
+   for (Map.Entry e : 
additionalTypes.entrySet()) {
+   addTypeMap.put(e.getKey(), 
TypeExtractor.createTypeInfo(e.getValue()));
+   }
+   rowTypeInfo = new RowTypeInfo(typeInfo, size, 
addTypeMap);
+   } else {
+   rowTypeInfo = new RowTypeInfo(typeInfo, size);
+   }
+
+   CsvInputFormat inputFormat = new RowCsvInputFormat(path, 
rowTypeInfo, this.includedMask);
+   configureInputFormat(inputFormat);
+
+   return new DataSource(executionContext, inputFormat, 
rowTypeInfo, Utils.getCallLocationName());
+   }
+
+   public DataSource rowType(Class mainTargetType, int size) {
--- End diff --

Please add javadoc


---
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 #3012: [FLINK-2186] Add readCsvAsRow methods to CsvReader...

[ex00]

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

https://github.com/apache/flink/pull/3012#discussion_r100247899
  
--- Diff: 
flink-tests/src/test/java/org/apache/flink/test/io/CsvReaderITCase.java ---
@@ -122,6 +127,80 @@ public void testValueTypes() throws Exception {
compareResultAsTuples(result, expected);
}
 
+   private int fullRowSize = 29;
+   private String including = "011";
+   private String fileContent =
+   "1,2,3," + 4 + "," + 5.0d + "," + true +
+   ",7,8,9,11,22,33,44,55,66,77,88,99,00," +
+   "111,222,333,444,555,666,777,888,999,000\n" +
+   "a,b,c," + 40 + "," + 50.0d + "," + false +
+   ",g,h,i,aa,bb,cc,dd,ee,ff,gg,hh,ii,mm," +
+   "aaa,bbb,ccc,ddd,eee,fff,ggg,hhh,iii,mmm\n";
--- End diff --

A lot of concatenations, the fileContent equals 
```
"1,2,3,4,5.0,true" +
",7,8,9,11,22,33,44,55,66,77,88,99,00," +
"111,222,333,444,555,666,777,888,999,000\n" +
"a,b,c,40,50.0,false," +
"g,h,i,aa,bb,cc,dd,ee,ff,gg,hh,ii,mm," +
"aaa,bbb,ccc,ddd,eee,fff,ggg,hhh,iii,mmm\n"
```


---
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-2186) Rework CSV import to support very wide files

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/3012#discussion_r100247870
  
--- Diff: 
flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
 ---
@@ -348,6 +349,47 @@ class ExecutionEnvironment(javaEnv: JavaEnv) {
 wrap(new DataSource[T](javaEnv, inputFormat, typeInfo, 
getCallLocationName()))
   }
 
+  def readCsvFileAsRow[T : ClassTag : TypeInformation](
--- End diff --

Could you add scaladoc for method?  may do not understand, what is 
```additionalTypes```


> Rework CSV import to support very wide files
> 
>
> Key: FLINK-2186
> URL: https://issues.apache.org/jira/browse/FLINK-2186
> Project: Flink
>  Issue Type: Improvement
>  Components: Machine Learning Library, Scala API
>Reporter: Theodore Vasiloudis
>Assignee: Anton Solovev
>
> In the current readVcsFile implementation, importing CSV files with many 
> columns can become from cumbersome to impossible.
> For example to import an 11 column file we need to write:
> {code}
> val cancer = env.readCsvFile[(String, String, String, String, String, String, 
> String, String, String, String, 
> String)]("/path/to/breast-cancer-wisconsin.data")
> {code}
> For many use cases in Machine Learning we might have CSV files with thousands 
> or millions of columns that we want to import as vectors.
> In that case using the current readCsvFile method becomes impossible.
> We therefore need to rework the current function, or create a new one that 
> will allow us to import CSV files with an arbitrary number of columns.



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

[GitHub] flink pull request #3012: [FLINK-2186] Add readCsvAsRow methods to CsvReader...

[ex00]

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

https://github.com/apache/flink/pull/3012#discussion_r100247879
  
--- Diff: 
flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
 ---
@@ -348,6 +349,47 @@ class ExecutionEnvironment(javaEnv: JavaEnv) {
 wrap(new DataSource[T](javaEnv, inputFormat, typeInfo, 
getCallLocationName()))
   }
 
+  def readCsvFileAsRow[T : ClassTag : TypeInformation](
+filePath: String,
+rowSize: Int,
+additionalTypes: Map[Int, Class[_]] = null,
+lineDelimiter: String = "\n",
+fieldDelimiter: String = ",",
+quoteCharacter: Character = null,
+ignoreFirstLine: Boolean = false,
+ignoreComments: String = null,
+lenient: Boolean = false,
+includedFields: Array[Int] = null): DataSet[Row] = {
--- End diff --

Please add more two spase for parameters like as for other methods in 
class, space chars should is 6.


---
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-2186) Rework CSV import to support very wide files

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/3012#discussion_r100247839
  
--- Diff: 
flink-core/src/main/java/org/apache/flink/api/java/typeutils/RowTypeInfo.java 
---
@@ -192,6 +193,28 @@ public void getFlatFields(String fieldExpression, int 
offset, List mainType, int size, Map additionalTypes) {
+   this(configureTypes(mainType, size, additionalTypes));
+   }
+
+   public RowTypeInfo(TypeInformation mainType, int size) {
+   this(configureTypes(mainType, size, Collections.emptyMap()));
+   }
+
+   private static TypeInformation[] configureTypes(TypeInformation 
mainType, int size, Map additionalTypes) {
--- End diff --

Could you format argumets like as for ```RowTypeInfo#createComparator#219```
for example:
```
private static TypeInformation[] configureTypes(
TypeInformation mainType,
int size,
Map additionalTypes) {
```


> Rework CSV import to support very wide files
> 
>
> Key: FLINK-2186
> URL: https://issues.apache.org/jira/browse/FLINK-2186
> Project: Flink
>  Issue Type: Improvement
>  Components: Machine Learning Library, Scala API
>Reporter: Theodore Vasiloudis
>Assignee: Anton Solovev
>
> In the current readVcsFile implementation, importing CSV files with many 
> columns can become from cumbersome to impossible.
> For example to import an 11 column file we need to write:
> {code}
> val cancer = env.readCsvFile[(String, String, String, String, String, String, 
> String, String, String, String, 
> String)]("/path/to/breast-cancer-wisconsin.data")
> {code}
> For many use cases in Machine Learning we might have CSV files with thousands 
> or millions of columns that we want to import as vectors.
> In that case using the current readCsvFile method becomes impossible.
> We therefore need to rework the current function, or create a new one that 
> will allow us to import CSV files with an arbitrary number of columns.



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

[jira] [Commented] (FLINK-2186) Rework CSV import to support very wide files

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/3012#discussion_r100247845
  
--- Diff: 
flink-java/src/main/java/org/apache/flink/api/java/io/CsvReader.java ---
@@ -351,6 +356,32 @@ public CsvReader ignoreInvalidLines(){
return new DataSource(executionContext, inputFormat, 
typeInfo, Utils.getCallLocationName());
}

+   public DataSource rowType(Class mainTargetType, int size, 
Map additionalTypes) {
--- End diff --

Please add javadoc


> Rework CSV import to support very wide files
> 
>
> Key: FLINK-2186
> URL: https://issues.apache.org/jira/browse/FLINK-2186
> Project: Flink
>  Issue Type: Improvement
>  Components: Machine Learning Library, Scala API
>Reporter: Theodore Vasiloudis
>Assignee: Anton Solovev
>
> In the current readVcsFile implementation, importing CSV files with many 
> columns can become from cumbersome to impossible.
> For example to import an 11 column file we need to write:
> {code}
> val cancer = env.readCsvFile[(String, String, String, String, String, String, 
> String, String, String, String, 
> String)]("/path/to/breast-cancer-wisconsin.data")
> {code}
> For many use cases in Machine Learning we might have CSV files with thousands 
> or millions of columns that we want to import as vectors.
> In that case using the current readCsvFile method becomes impossible.
> We therefore need to rework the current function, or create a new one that 
> will allow us to import CSV files with an arbitrary number of columns.



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

[jira] [Commented] (FLINK-2186) Rework CSV import to support very wide files

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/3012#discussion_r100247856
  
--- Diff: 
flink-java/src/main/java/org/apache/flink/api/java/io/CsvReader.java ---
@@ -351,6 +356,32 @@ public CsvReader ignoreInvalidLines(){
return new DataSource(executionContext, inputFormat, 
typeInfo, Utils.getCallLocationName());
}

+   public DataSource rowType(Class mainTargetType, int size, 
Map additionalTypes) {
+   Preconditions.checkNotNull(mainTargetType, "The main type class 
must not be null.");
+
+   TypeInformation typeInfo = 
TypeExtractor.createTypeInfo(mainTargetType);
+   RowTypeInfo rowTypeInfo;
+
+   if (additionalTypes != null) {
+   Map addTypeMap = new 
HashMap<>(additionalTypes.size());
+   for (Map.Entry e : 
additionalTypes.entrySet()) {
+   addTypeMap.put(e.getKey(), 
TypeExtractor.createTypeInfo(e.getValue()));
+   }
+   rowTypeInfo = new RowTypeInfo(typeInfo, size, 
addTypeMap);
+   } else {
+   rowTypeInfo = new RowTypeInfo(typeInfo, size);
+   }
+
+   CsvInputFormat inputFormat = new RowCsvInputFormat(path, 
rowTypeInfo, this.includedMask);
+   configureInputFormat(inputFormat);
+
+   return new DataSource(executionContext, inputFormat, 
rowTypeInfo, Utils.getCallLocationName());
+   }
+
+   public DataSource rowType(Class mainTargetType, int size) {
--- End diff --

Please add javadoc


> Rework CSV import to support very wide files
> 
>
> Key: FLINK-2186
> URL: https://issues.apache.org/jira/browse/FLINK-2186
> Project: Flink
>  Issue Type: Improvement
>  Components: Machine Learning Library, Scala API
>Reporter: Theodore Vasiloudis
>Assignee: Anton Solovev
>
> In the current readVcsFile implementation, importing CSV files with many 
> columns can become from cumbersome to impossible.
> For example to import an 11 column file we need to write:
> {code}
> val cancer = env.readCsvFile[(String, String, String, String, String, String, 
> String, String, String, String, 
> String)]("/path/to/breast-cancer-wisconsin.data")
> {code}
> For many use cases in Machine Learning we might have CSV files with thousands 
> or millions of columns that we want to import as vectors.
> In that case using the current readCsvFile method becomes impossible.
> We therefore need to rework the current function, or create a new one that 
> will allow us to import CSV files with an arbitrary number of columns.



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

[GitHub] flink pull request #3012: [FLINK-2186] Add readCsvAsRow methods to CsvReader...

[ex00]

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

https://github.com/apache/flink/pull/3012#discussion_r100247870
  
--- Diff: 
flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
 ---
@@ -348,6 +349,47 @@ class ExecutionEnvironment(javaEnv: JavaEnv) {
 wrap(new DataSource[T](javaEnv, inputFormat, typeInfo, 
getCallLocationName()))
   }
 
+  def readCsvFileAsRow[T : ClassTag : TypeInformation](
--- End diff --

Could you add scaladoc for method?  may do not understand, what is 
```additionalTypes```


---
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 #3012: [FLINK-2186] Add readCsvAsRow methods to CsvReader...

[ex00]

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

https://github.com/apache/flink/pull/3012#discussion_r100247839
  
--- Diff: 
flink-core/src/main/java/org/apache/flink/api/java/typeutils/RowTypeInfo.java 
---
@@ -192,6 +193,28 @@ public void getFlatFields(String fieldExpression, int 
offset, List mainType, int size, Map additionalTypes) {
+   this(configureTypes(mainType, size, additionalTypes));
+   }
+
+   public RowTypeInfo(TypeInformation mainType, int size) {
+   this(configureTypes(mainType, size, Collections.emptyMap()));
+   }
+
+   private static TypeInformation[] configureTypes(TypeInformation 
mainType, int size, Map additionalTypes) {
--- End diff --

Could you format argumets like as for ```RowTypeInfo#createComparator#219```
for example:
```
private static TypeInformation[] configureTypes(
TypeInformation mainType,
int size,
Map additionalTypes) {
```


---
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 #3012: [FLINK-2186] Add readCsvAsRow methods to CsvReader and sc...

[ex00]

Github user ex00 commented on the issue:

https://github.com/apache/flink/pull/3012
  
Thanks for work  @tonycox.
The PR looks good to me. I left a few minor comments about PR.

What do you mean about test negative case? If typeMap does not match with 
fields type in file for example


---
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] [Assigned] (FLINK-5658) Add event time OVER RANGE BETWEEN UNBOUNDED PRECEDING aggregation to SQL

[Yuhong Hong (JIRA)]

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

Yuhong Hong reassigned FLINK-5658:
--

Assignee: Yuhong Hong  (was: sunjincheng)

> Add event time OVER RANGE BETWEEN UNBOUNDED PRECEDING aggregation to SQL
> 
>
> Key: FLINK-5658
> URL: https://issues.apache.org/jira/browse/FLINK-5658
> Project: Flink
>  Issue Type: Sub-task
>  Components: Table API & SQL
>Reporter: Fabian Hueske
>Assignee: Yuhong Hong
>
> The goal of this issue is to add support for OVER RANGE aggregations on event 
> time streams to the SQL interface.
> Queries similar to the following should be supported:
> {code}
> SELECT 
>   a, 
>   SUM(b) OVER (PARTITION BY c ORDER BY rowTime() RANGE BETWEEN UNBOUNDED 
> PRECEDING AND CURRENT ROW) AS sumB,
>   MIN(b) OVER (PARTITION BY c ORDER BY rowTime() RANGE BETWEEN UNBOUNDED 
> PRECEDING AND CURRENT ROW) AS minB
> FROM myStream
> {code}
> The following restrictions should initially apply:
> - All OVER clauses in the same SELECT clause must be exactly the same.
> - The PARTITION BY clause is optional (no partitioning results in single 
> threaded execution).
> - The ORDER BY clause may only have rowTime() as parameter. rowTime() is a 
> parameterless scalar function that just indicates processing time mode.
> - bounded PRECEDING is not supported (see FLINK-5655)
> - FOLLOWING is not supported.
> The restrictions will be resolved in follow up issues. If we find that some 
> of the restrictions are trivial to address, we can add the functionality in 
> this issue as well.
> This issue includes:
> - Design of the DataStream operator to compute OVER ROW aggregates
> - Translation from Calcite's RelNode representation (LogicalProject with 
> RexOver expression).



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

[jira] [Commented] (FLINK-5749) unset HADOOP_HOME and HADOOP_CONF_DIR to avoid env in build machine failing the UT and IT

[ASF GitHub Bot (JIRA)]

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

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

GitHub user wenlong88 opened a pull request:

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

[FLINK-5749]unset HADOOP_HOME and HADOOP_CONF_DIR to avoid env in build 
machine failing the UT and IT

Currently when we are trying to build flink on a machine with HADOOP_HOME 
environment variable set, Test data will be written to HDFS, instead of local 
tmp dir which is expected. This will cause tests failed. 
This PR unsets HADOOP_HOME and HADOOP_CONF_DIR environment variable in pom, 
to make sure maven run the test cases in the exactly way we want.

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

$ git pull https://github.com/wenlong88/flink jira-5749

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

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


commit d0e6464f37e2057c508a0e7858d3b42778610637
Author: wenlong.lwl 
Date:   2017-01-13T06:12:17Z

[to #9896513] Unset HADOOP_HOME and HADOOP_CONF_DIR in pom

Summary:
unset HADOOP_HOME and HADOOP_CONF_DIR to avoid env in build machine failing 
the UT and IT

fix T9896513

Test Plan: UT/IT

Reviewers: 黎钢

Reviewed By: 黎钢

Subscribers: P577102

Differential Revision: https://aone.alibaba-inc.com/code/D96861




> unset HADOOP_HOME and HADOOP_CONF_DIR to avoid env in build machine 
> failing the UT and IT
> -
>
> Key: FLINK-5749
> URL: https://issues.apache.org/jira/browse/FLINK-5749
> Project: Flink
>  Issue Type: Bug
>Reporter: Wenlong Lyu
>Assignee: Wenlong Lyu
>
> Currently when we are trying to build flink on a machine with HADOOP_HOME 
> environment variable set, Test data will be written to HDFS, instead of local 
> tmp dir which is expected. This will cause tests failed. 
> I suggest unset HADOOP_HOME and HADOOP_CONF_DIR environment variable in pom, 
> to make sure maven run the test cases in the exactly way we want.



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

[GitHub] flink pull request #3288: [FLINK-5749]unset HADOOP_HOME and HADOOP_CONF_DIR ...

[wenlong88]

GitHub user wenlong88 opened a pull request:

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

[FLINK-5749]unset HADOOP_HOME and HADOOP_CONF_DIR to avoid env in build 
machine failing the UT and IT

Currently when we are trying to build flink on a machine with HADOOP_HOME 
environment variable set, Test data will be written to HDFS, instead of local 
tmp dir which is expected. This will cause tests failed. 
This PR unsets HADOOP_HOME and HADOOP_CONF_DIR environment variable in pom, 
to make sure maven run the test cases in the exactly way we want.

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

$ git pull https://github.com/wenlong88/flink jira-5749

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

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


commit d0e6464f37e2057c508a0e7858d3b42778610637
Author: wenlong.lwl 
Date:   2017-01-13T06:12:17Z

[to #9896513] Unset HADOOP_HOME and HADOOP_CONF_DIR in pom

Summary:
unset HADOOP_HOME and HADOOP_CONF_DIR to avoid env in build machine failing 
the UT and IT

fix T9896513

Test Plan: UT/IT

Reviewers: 黎钢

Reviewed By: 黎钢

Subscribers: P577102

Differential Revision: https://aone.alibaba-inc.com/code/D96861




---
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-5749) unset HADOOP_HOME and HADOOP_CONF_DIR to avoid env in build machine failing the UT and IT

[Wenlong Lyu (JIRA)]

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

Wenlong Lyu updated FLINK-5749:
---
Summary: unset HADOOP_HOME and HADOOP_CONF_DIR to avoid env in build 
machine failing the UT and IT  (was: Unset HADOOP_HOME and HADOOP_CONF_DIR from 
system environment when building flink)

> unset HADOOP_HOME and HADOOP_CONF_DIR to avoid env in build machine 
> failing the UT and IT
> -
>
> Key: FLINK-5749
> URL: https://issues.apache.org/jira/browse/FLINK-5749
> Project: Flink
>  Issue Type: Bug
>Reporter: Wenlong Lyu
>Assignee: Wenlong Lyu
>
> Currently when we are trying to build flink on a machine with HADOOP_HOME 
> environment variable set, Test data will be written to HDFS, instead of local 
> tmp dir which is expected. This will cause tests failed. 
> I suggest unset HADOOP_HOME and HADOOP_CONF_DIR environment variable in pom, 
> to make sure maven run the test cases in the exactly way we want.



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

[jira] [Commented] (FLINK-1731) Add kMeans clustering algorithm to machine learning library

[ASF GitHub Bot (JIRA)]

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

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

Github user sachingoel0101 commented on the issue:

https://github.com/apache/flink/pull/3192
  
@skonto I'm traveling right now and won't be able to push an update until
Monday/Tuesday.

On Feb 9, 2017 09:31, "Stavros Kontopoulos" 
wrote:

> @sachingoel0101  could you update the
> PR so I can do a final review and request a merge?
> @tillrohrmann  could assist with the
> forwardedfields question?
>
> —
> You are receiving this because you were mentioned.
> Reply to this email directly, view it on GitHub
> , or 
mute
> the thread
> 

> .
>



> Add kMeans clustering algorithm to machine learning library
> ---
>
> Key: FLINK-1731
> URL: https://issues.apache.org/jira/browse/FLINK-1731
> Project: Flink
>  Issue Type: New Feature
>  Components: Machine Learning Library
>Reporter: Till Rohrmann
>Assignee: Peter Schrott
>  Labels: ML
>
> The Flink repository already contains a kMeans implementation but it is not 
> yet ported to the machine learning library. I assume that only the used data 
> types have to be adapted and then it can be more or less directly moved to 
> flink-ml.
> The kMeans++ [1] and the kMeans|| [2] algorithm constitute a better 
> implementation because the improve the initial seeding phase to achieve near 
> optimal clustering. It might be worthwhile to implement kMeans||.
> Resources:
> [1] http://ilpubs.stanford.edu:8090/778/1/2006-13.pdf
> [2] http://theory.stanford.edu/~sergei/papers/vldb12-kmpar.pdf



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

[GitHub] flink issue #3192: [FLINK-1731][ml] Add KMeans clustering(Lloyd's algorithm)

[sachingoel0101]

Github user sachingoel0101 commented on the issue:

https://github.com/apache/flink/pull/3192
  
@skonto I'm traveling right now and won't be able to push an update until
Monday/Tuesday.

On Feb 9, 2017 09:31, "Stavros Kontopoulos" 
wrote:

> @sachingoel0101  could you update the
> PR so I can do a final review and request a merge?
> @tillrohrmann  could assist with the
> forwardedfields question?
>
> —
> You are receiving this because you were mentioned.
> Reply to this email directly, view it on GitHub
> , or 
mute
> the thread
> 

> .
>



---
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-5517) Upgrade hbase version to 1.3.0

[ASF GitHub Bot (JIRA)]

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

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

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/3235
  
Merging to `master`


> Upgrade hbase version to 1.3.0
> --
>
> Key: FLINK-5517
> URL: https://issues.apache.org/jira/browse/FLINK-5517
> Project: Flink
>  Issue Type: Improvement
>  Components: Streaming Connectors
>Reporter: Ted Yu
>
> In the thread 'Help using HBase with Flink 1.1.4', Giuliano reported seeing:
> {code}
> java.lang.IllegalAccessError: tried to access method 
> com.google.common.base.Stopwatch.()V from class 
> org.apache.hadoop.hbase.zookeeper.MetaTableLocator
> {code}
> The above has been solved by HBASE-14963
> hbase 1.3.0 is being released.
> We should upgrade hbase dependency to 1.3.0



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

[GitHub] flink issue #3235: FLINK-5517 Upgrade hbase version to 1.3.0

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/3235
  
Merging to `master`


---
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-5564) User Defined Aggregates

[Shaoxuan Wang (JIRA)]

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

Shaoxuan Wang edited comment on FLINK-5564 at 2/9/17 5:16 AM:
--

Thanks [~fhueske], 
Absolutely, I agree with you that it is better to separate the huge PR into 
some ones. (merge 1,2,3 will lead to more than 3K lines change)
But I am afraid I did not completely get your suggested #1. Migrating the 
existing Agg without changing runtime code will lead to all Integration Test 
fail. One possible way is that I create new interface (say AggregateFunction) 
and create a few Aggs which is implemented from new interface (say intAgg 
extends AggregateFunction), and in step #1, I just add queryPlan tests, like 
what we usually did in GroupWindowTest. Is this what you are suggesting.


was (Author: shaoxuanwang):
Thanks [~fhueske], 
Absolutely, I agree with you that it is better to separate the huge PR into 
some ones. (merge 1,2,3 will lead to more than 3K lines change)
But I am afraid I did not completely get your suggested #1. Migrating the 
existing Agg without changing runtime code will lead to all IntergrationTest 
fail. One possible way is that I create new interface (say AggregateFunction) 
and create a few Aggs which is implemented from new interface (say intAgg 
extends AggregateFunction), and in step #1, I just add queryPlan tests, like 
what we usually did in GroupWindowTest. Is this what you are suggesting.

> User Defined Aggregates
> ---
>
> Key: FLINK-5564
> URL: https://issues.apache.org/jira/browse/FLINK-5564
> Project: Flink
>  Issue Type: Improvement
>  Components: Table API & SQL
>Reporter: Shaoxuan Wang
>Assignee: Shaoxuan Wang
>
> User-defined aggregates would be a great addition to the Table API / SQL.
> The current aggregate interface is not well suited for the external users.  
> This issue proposes to redesign the aggregate such that we can expose an 
> better external UDAGG interface to the users. The detailed design proposal 
> can be found here: 
> https://docs.google.com/document/d/19JXK8jLIi8IqV9yf7hOs_Oz67yXOypY7Uh5gIOK2r-U/edit
> Motivation:
> 1. The current aggregate interface is not very concise to the users. One 
> needs to know the design details of the intermediate Row buffer before 
> implements an Aggregate. Seven functions are needed even for a simple Count 
> aggregate.
> 2. Another limitation of current aggregate function is that it can only be 
> applied on one single column. There are many scenarios which require the 
> aggregate function taking multiple columns as the inputs.
> 3. “Retraction” is not considered and covered in the current Aggregate.
> 4. It might be very good to have a local/global aggregate query plan 
> optimization, which is very promising to optimize UDAGG performance in some 
> scenarios.
> Proposed Changes:
> 1. Implement an aggregate dataStream API (Done by 
> [FLINK-5582|https://issues.apache.org/jira/browse/FLINK-5582])
> 2. Update all the existing aggregates to use the new aggregate dataStream API
> 3. Provide a better User-Defined Aggregate interface
> 4. Add retraction support
> 5. Add local/global aggregate



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

[jira] [Comment Edited] (FLINK-5564) User Defined Aggregates

[Shaoxuan Wang (JIRA)]

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

Shaoxuan Wang edited comment on FLINK-5564 at 2/9/17 5:16 AM:
--

Thanks [~fhueske], 
Absolutely, I agree with you that it is better to separate the huge PR into 
some ones. (merge 1,2,3 will lead to more than 3K lines change)
But I am afraid I did not completely get your suggested #1. Migrating the 
existing Agg without changing runtime code will lead to all IntergrationTest 
fail. One possible way is that I create new interface (say AggregateFunction) 
and create a few Aggs which is implemented from new interface (say intAgg 
extends AggregateFunction), and in step #1, I just add queryPlan tests, like 
what we usually did in GroupWindowTest. Is this what you are suggesting.


was (Author: shaoxuanwang):
Thanks [~fhueske], 
Obsoletely, I agree with you that it is better to separate the huge PR into 
some ones. (merge 1,2,3 will lead to more than 3K lines change)
But I am afraid I did not completely get your suggested #1. Migrating the 
existing Agg without changing runtime code will lead to all IntergrationTest 
fail. One possible way is that I create new interface (say AggregateFunction) 
and create a few Aggs which is implemented from new interface (say intAgg 
extends AggregateFunction), and in step #1, I just add queryPlan tests, like 
what we usually did in GroupWindowTest. Is this what you are suggesting.

> User Defined Aggregates
> ---
>
> Key: FLINK-5564
> URL: https://issues.apache.org/jira/browse/FLINK-5564
> Project: Flink
>  Issue Type: Improvement
>  Components: Table API & SQL
>Reporter: Shaoxuan Wang
>Assignee: Shaoxuan Wang
>
> User-defined aggregates would be a great addition to the Table API / SQL.
> The current aggregate interface is not well suited for the external users.  
> This issue proposes to redesign the aggregate such that we can expose an 
> better external UDAGG interface to the users. The detailed design proposal 
> can be found here: 
> https://docs.google.com/document/d/19JXK8jLIi8IqV9yf7hOs_Oz67yXOypY7Uh5gIOK2r-U/edit
> Motivation:
> 1. The current aggregate interface is not very concise to the users. One 
> needs to know the design details of the intermediate Row buffer before 
> implements an Aggregate. Seven functions are needed even for a simple Count 
> aggregate.
> 2. Another limitation of current aggregate function is that it can only be 
> applied on one single column. There are many scenarios which require the 
> aggregate function taking multiple columns as the inputs.
> 3. “Retraction” is not considered and covered in the current Aggregate.
> 4. It might be very good to have a local/global aggregate query plan 
> optimization, which is very promising to optimize UDAGG performance in some 
> scenarios.
> Proposed Changes:
> 1. Implement an aggregate dataStream API (Done by 
> [FLINK-5582|https://issues.apache.org/jira/browse/FLINK-5582])
> 2. Update all the existing aggregates to use the new aggregate dataStream API
> 3. Provide a better User-Defined Aggregate interface
> 4. Add retraction support
> 5. Add local/global aggregate



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

[jira] [Commented] (FLINK-5564) User Defined Aggregates

[Shaoxuan Wang (JIRA)]

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

Shaoxuan Wang commented on FLINK-5564:
--

Thanks [~fhueske], 
Obsoletely, I agree with you that it is better to separate the huge PR into 
some ones. (merge 1,2,3 will lead to more than 3K lines change)
But I am afraid I did not completely get your suggested #1. Migrating the 
existing Agg without changing runtime code will lead to all IntergrationTest 
fail. One possible way is that I create new interface (say AggregateFunction) 
and create a few Aggs which is implemented from new interface (say intAgg 
extends AggregateFunction), and in step #1, I just add queryPlan tests, like 
what we usually did in GroupWindowTest. Is this what you are suggesting.

> User Defined Aggregates
> ---
>
> Key: FLINK-5564
> URL: https://issues.apache.org/jira/browse/FLINK-5564
> Project: Flink
>  Issue Type: Improvement
>  Components: Table API & SQL
>Reporter: Shaoxuan Wang
>Assignee: Shaoxuan Wang
>
> User-defined aggregates would be a great addition to the Table API / SQL.
> The current aggregate interface is not well suited for the external users.  
> This issue proposes to redesign the aggregate such that we can expose an 
> better external UDAGG interface to the users. The detailed design proposal 
> can be found here: 
> https://docs.google.com/document/d/19JXK8jLIi8IqV9yf7hOs_Oz67yXOypY7Uh5gIOK2r-U/edit
> Motivation:
> 1. The current aggregate interface is not very concise to the users. One 
> needs to know the design details of the intermediate Row buffer before 
> implements an Aggregate. Seven functions are needed even for a simple Count 
> aggregate.
> 2. Another limitation of current aggregate function is that it can only be 
> applied on one single column. There are many scenarios which require the 
> aggregate function taking multiple columns as the inputs.
> 3. “Retraction” is not considered and covered in the current Aggregate.
> 4. It might be very good to have a local/global aggregate query plan 
> optimization, which is very promising to optimize UDAGG performance in some 
> scenarios.
> Proposed Changes:
> 1. Implement an aggregate dataStream API (Done by 
> [FLINK-5582|https://issues.apache.org/jira/browse/FLINK-5582])
> 2. Update all the existing aggregates to use the new aggregate dataStream API
> 3. Provide a better User-Defined Aggregate interface
> 4. Add retraction support
> 5. Add local/global aggregate



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

[jira] [Commented] (FLINK-1731) Add kMeans clustering algorithm to machine learning library

[ASF GitHub Bot (JIRA)]

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

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

Github user skonto commented on the issue:

https://github.com/apache/flink/pull/3192
  
@sachingoel0101 could you update the PR so I can do a final review and 
request a merge?
@tillrohrmann could assist with the forwardedfields question?


> Add kMeans clustering algorithm to machine learning library
> ---
>
> Key: FLINK-1731
> URL: https://issues.apache.org/jira/browse/FLINK-1731
> Project: Flink
>  Issue Type: New Feature
>  Components: Machine Learning Library
>Reporter: Till Rohrmann
>Assignee: Peter Schrott
>  Labels: ML
>
> The Flink repository already contains a kMeans implementation but it is not 
> yet ported to the machine learning library. I assume that only the used data 
> types have to be adapted and then it can be more or less directly moved to 
> flink-ml.
> The kMeans++ [1] and the kMeans|| [2] algorithm constitute a better 
> implementation because the improve the initial seeding phase to achieve near 
> optimal clustering. It might be worthwhile to implement kMeans||.
> Resources:
> [1] http://ilpubs.stanford.edu:8090/778/1/2006-13.pdf
> [2] http://theory.stanford.edu/~sergei/papers/vldb12-kmpar.pdf



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

[GitHub] flink issue #3192: [FLINK-1731][ml] Add KMeans clustering(Lloyd's algorithm)

[skonto]

Github user skonto commented on the issue:

https://github.com/apache/flink/pull/3192
  
@sachingoel0101 could you update the PR so I can do a final review and 
request a merge?
@tillrohrmann could assist with the forwardedfields question?


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

[jira] [Created] (FLINK-5749) Unset HADOOP_HOME and HADOOP_CONF_DIR from system environment when building flink

[Wenlong Lyu (JIRA)]

Wenlong Lyu created FLINK-5749:
--

 Summary: Unset HADOOP_HOME and HADOOP_CONF_DIR from system 
environment when building flink
 Key: FLINK-5749
 URL: https://issues.apache.org/jira/browse/FLINK-5749
 Project: Flink
  Issue Type: Bug
Reporter: Wenlong Lyu
Assignee: Wenlong Lyu


Currently when we are trying to build flink on a machine with HADOOP_HOME 
environment variable set, Test data will be written to HDFS, instead of local 
tmp dir which is expected. This will cause tests failed. 
I suggest unset HADOOP_HOME and HADOOP_CONF_DIR environment variable in pom, to 
make sure maven run the test cases in the exactly way we want.



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

[GitHub] flink pull request #3151: [FLINK-4364][runtime]mplement TaskManager side of ...

[wangzhijiang999]

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

https://github.com/apache/flink/pull/3151#discussion_r100222823
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
 ---
@@ -112,6 +114,13 @@ public TaskManagerRunner(
// Initialize the TM metrics

TaskExecutorMetricsInitializer.instantiateStatusMetrics(taskManagerMetricGroup, 
taskManagerServices.getNetworkEnvironment());
 
+   HeartbeatManagerImpl heartbeatManager = new 
HeartbeatManagerImpl(
+   
taskManagerConfiguration.getTimeout().toMilliseconds(),
+   resourceID,
+   executor,
+   Executors.newSingleThreadScheduledExecutor(),
--- End diff --

Yes, I totally agree with that and the advantages can avoid bringing extra 
thread pool and run all the PRC messages in the uniform executor if 
**RPCService** can handle well.


---
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-4364) Implement TaskManager side of heartbeat from JobManager

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/3151#discussion_r100222823
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
 ---
@@ -112,6 +114,13 @@ public TaskManagerRunner(
// Initialize the TM metrics

TaskExecutorMetricsInitializer.instantiateStatusMetrics(taskManagerMetricGroup, 
taskManagerServices.getNetworkEnvironment());
 
+   HeartbeatManagerImpl heartbeatManager = new 
HeartbeatManagerImpl(
+   
taskManagerConfiguration.getTimeout().toMilliseconds(),
+   resourceID,
+   executor,
+   Executors.newSingleThreadScheduledExecutor(),
--- End diff --

Yes, I totally agree with that and the advantages can avoid bringing extra 
thread pool and run all the PRC messages in the uniform executor if 
**RPCService** can handle well.


> Implement TaskManager side of heartbeat from JobManager
> ---
>
> Key: FLINK-4364
> URL: https://issues.apache.org/jira/browse/FLINK-4364
> Project: Flink
>  Issue Type: Sub-task
>  Components: Cluster Management
>Reporter: Zhijiang Wang
>Assignee: Zhijiang Wang
>
> The {{JobManager}} initiates heartbeat messages via (JobID, JmLeaderID), and 
> the {{TaskManager}} will report metrics info for each heartbeat.



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

[jira] [Commented] (FLINK-2184) Cannot get last element with maxBy/minBy

[John Mu (JIRA)]

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

John Mu commented on FLINK-2184:


Is anyone working on this? The PR looks like it got messed up.
I can have a go.

However, if the goal is for the Java and Scala APIs to be the same there is an 
issue. 
The default value on {{first}} for {{ComparableAggregator}} is false. This true 
for the {{min}} and {{max}} aggregators in the Java API. However, for the Scala 
API {{org.apache.flink.streaming.api.scala.WindowedStream#aggregate}} sets 
{{first}} to true by default. Which one should it be?
I'm guessing the Scala method needs more cases for the {{max}} and {{min}} 
case. 



> Cannot get last element with maxBy/minBy
> 
>
> Key: FLINK-2184
> URL: https://issues.apache.org/jira/browse/FLINK-2184
> Project: Flink
>  Issue Type: Improvement
>  Components: Scala API, Streaming
>Reporter: Gábor Hermann
>Priority: Minor
>
> In the streaming Scala API there is no method
> {{maxBy(int positionToMaxBy, boolean first)}}
> nor
> {{minBy(int positionToMinBy, boolean first)}}
> like in the Java API, where _first_ set to {{true}} indicates that the latest 
> found element will return.
> These methods should be added to the Scala API too, in order to be consistent.



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

[jira] [Comment Edited] (FLINK-5679) Refactor *CheckpointedITCase tests to speed up

[Andrew Efimov (JIRA)]

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

Andrew Efimov edited comment on FLINK-5679 at 2/9/17 1:59 AM:
--

Current status: I am collected all tests related to testing stream topology and 
checkpointing in {{StreamCheckpointingITCase}}, which will be executed on the 
same cluster.
in the current portion of refactoring left to finish these two tests: 
{{PartitionedStateCheckpointingITCase}} 
{{UdfStreamOperatorCheckpointingITCase}}.

It is sad that the tests are not up to date, and focus and control over them 
are lost. We should use more conscientious approach to testing.
I intend to bring order in the part of checkpointing.



was (Author: andrew efimov):
Current status: I am collected all tests related to testing stream topology and 
checkpointing in {{StreamCheckpointingITCase}}.
in the current portion of refactoring left to finish these two tests: 
{{PartitionedStateCheckpointingITCase}} 
{{UdfStreamOperatorCheckpointingITCase}}.

It is sad that the tests are not up to date, and focus and control over them 
are lost. We should use more conscientious approach to testing.
I intend to bring order in the part of checkpointing.


> Refactor  *CheckpointedITCase tests to speed up
> ---
>
> Key: FLINK-5679
> URL: https://issues.apache.org/jira/browse/FLINK-5679
> Project: Flink
>  Issue Type: Test
>  Components: Tests
>Reporter: Andrew Efimov
>Assignee: Andrew Efimov
>  Labels: test-framework
> Fix For: 1.3.0
>
>
> Tests refactoring to speed up:
> {noformat}
> Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 40.193 sec - 
> in org.apache.flink.test.checkpointing.StreamCheckpointingITCasee
> Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 119.063 sec - 
> in org.apache.flink.test.checkpointing.UdfStreamOperatorCheckpointingITCase
> Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 47.525 sec - 
> in org.apache.flink.test.checkpointing.PartitionedStateCheckpointingITCase
> Tests run: 5, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 40.355 sec - 
> in org.apache.flink.test.checkpointing.EventTimeAllWindowCheckpointingITCase
> Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 51.615 sec - 
> in org.apache.flink.test.checkpointing.StateCheckpointedITCase
> {noformat}
> Tests could be adjusted in a similar way to save some time (some may actually 
> even be redundant by now)  
> https://github.com/StephanEwen/incubator-flink/commit/0dd7ae693f30585283d334a1d65b3d8222b7ca5c



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

[jira] [Commented] (FLINK-5679) Refactor *CheckpointedITCase tests to speed up

[Andrew Efimov (JIRA)]

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

Andrew Efimov commented on FLINK-5679:
--

Current status: I am collected all tests related to testing stream topology and 
checkpointing in {{StreamCheckpointingITCase}}.
in the current portion of refactoring left to finish these two tests: 
{{PartitionedStateCheckpointingITCase}} 
{{UdfStreamOperatorCheckpointingITCase}}.

It is sad that the tests are not up to date, and focus and control over them 
are lost. We should use more conscientious approach to testing.
I intend to bring order in the part of checkpointing.


> Refactor  *CheckpointedITCase tests to speed up
> ---
>
> Key: FLINK-5679
> URL: https://issues.apache.org/jira/browse/FLINK-5679
> Project: Flink
>  Issue Type: Test
>  Components: Tests
>Reporter: Andrew Efimov
>Assignee: Andrew Efimov
>  Labels: test-framework
> Fix For: 1.3.0
>
>
> Tests refactoring to speed up:
> {noformat}
> Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 40.193 sec - 
> in org.apache.flink.test.checkpointing.StreamCheckpointingITCasee
> Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 119.063 sec - 
> in org.apache.flink.test.checkpointing.UdfStreamOperatorCheckpointingITCase
> Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 47.525 sec - 
> in org.apache.flink.test.checkpointing.PartitionedStateCheckpointingITCase
> Tests run: 5, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 40.355 sec - 
> in org.apache.flink.test.checkpointing.EventTimeAllWindowCheckpointingITCase
> Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 51.615 sec - 
> in org.apache.flink.test.checkpointing.StateCheckpointedITCase
> {noformat}
> Tests could be adjusted in a similar way to save some time (some may actually 
> even be redundant by now)  
> https://github.com/StephanEwen/incubator-flink/commit/0dd7ae693f30585283d334a1d65b3d8222b7ca5c



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

[jira] [Commented] (FLINK-5724) Error in the 'Zipping Elements' docs

[John Mu (JIRA)]

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

John Mu commented on FLINK-5724:


Are you referring to 
https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/batch/zip_elements_guide.html?

It looks fine to me.

> Error in the 'Zipping Elements' docs
> 
>
> Key: FLINK-5724
> URL: https://issues.apache.org/jira/browse/FLINK-5724
> Project: Flink
>  Issue Type: Bug
>Reporter: Fokko Driesprong
>
> The tab for the Python documentation isn't working because there are two tabs 
> pointing at the Scala example.



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

[jira] [Commented] (FLINK-5631) [yarn] Support downloading additional jars from non-HDFS paths

[Haohui Mai (JIRA)]

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

Haohui Mai commented on FLINK-5631:
---

[~StephanEwen] can you please take another look?

> [yarn] Support downloading additional jars from non-HDFS paths
> --
>
> Key: FLINK-5631
> URL: https://issues.apache.org/jira/browse/FLINK-5631
> Project: Flink
>  Issue Type: Bug
>  Components: YARN
>Reporter: Haohui Mai
>Assignee: Haohui Mai
>
> Currently the {{YarnResourceManager}} and {{YarnApplicationMasterRunner}} 
> always register the additional jars using the YARN filesystem object. This is 
> problematic as the paths might require another filesystem.
> To support localizing from non-HDFS paths (e.g., s3, http or viewfs), the 
> cleaner approach is to get the filesystem object from the path.



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

[GitHub] flink pull request #3205: [FLINK-5635] Improve Docker tooling to make it eas...

[tonycox]

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

https://github.com/apache/flink/pull/3205#discussion_r100159827
  
--- Diff: flink-contrib/docker-flink/docker-compose.yml ---
@@ -16,21 +16,22 @@
 # limitations under the License.
 

 
-version: "2"
+# Set the FLINK_DOCKER_IMAGE_NAME environment variable to override the 
image name to use
+
+version: "2.1"
 services:
   jobmanager:
-image: flink
-container_name: "jobmanager"
+image: ${FLINK_DOCKER_IMAGE_NAME:-flink}
 expose:
   - "6123"
--- End diff --

do we need this if it set in Dockerfile ?


---
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-5635) Improve Docker tooling to make it easier to build images and launch Flink via Docker tools

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/3205#discussion_r100159827
  
--- Diff: flink-contrib/docker-flink/docker-compose.yml ---
@@ -16,21 +16,22 @@
 # limitations under the License.
 

 
-version: "2"
+# Set the FLINK_DOCKER_IMAGE_NAME environment variable to override the 
image name to use
+
+version: "2.1"
 services:
   jobmanager:
-image: flink
-container_name: "jobmanager"
+image: ${FLINK_DOCKER_IMAGE_NAME:-flink}
 expose:
   - "6123"
--- End diff --

do we need this if it set in Dockerfile ?


> Improve Docker tooling to make it easier to build images and launch Flink via 
> Docker tools
> --
>
> Key: FLINK-5635
> URL: https://issues.apache.org/jira/browse/FLINK-5635
> Project: Flink
>  Issue Type: Improvement
>  Components: Docker
>Affects Versions: 1.2.0
>Reporter: Jamie Grier
>Assignee: Jamie Grier
>
> This is a bit of a catch-all ticket for general improvements to the Flink on 
> Docker experience.
> Things to improve:
>   - Make it possible to build a Docker image from your own flink-dist 
> directory as well as official releases.
>   - Make it possible to override the image name so a user can more easily 
> publish these images to their Docker repository
>   - Provide scripts that show how to properly run on Docker Swarm or similar 
> environments with overlay networking (Kubernetes) without using host 
> networking.
>   - Log to stdout rather than to files.
>   - Work properly with docker-compose for local deployment as well as 
> production deployments (Swarm/k8s)



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

[jira] [Commented] (FLINK-5635) Improve Docker tooling to make it easier to build images and launch Flink via Docker tools

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/3205#discussion_r100156082
  
--- Diff: flink-contrib/docker-flink/Dockerfile ---
@@ -36,22 +31,25 @@ ENV PATH $PATH:$FLINK_HOME/bin
 EXPOSE 8081
 EXPOSE 6123
 
+# flink-dist can point to a directory, a tarball on the local system, or a 
url to a tarball
+ARG flink_dist=NOT_SET
+
 # Install build dependencies and flink
+ADD $flink_dist $FLINK_INSTALL_PATH
 RUN set -x && \
   mkdir -p $FLINK_INSTALL_PATH && \
--- End diff --

and what if `$flink_dist` is url ? 
Should run `tar xzf` I think


> Improve Docker tooling to make it easier to build images and launch Flink via 
> Docker tools
> --
>
> Key: FLINK-5635
> URL: https://issues.apache.org/jira/browse/FLINK-5635
> Project: Flink
>  Issue Type: Improvement
>  Components: Docker
>Affects Versions: 1.2.0
>Reporter: Jamie Grier
>Assignee: Jamie Grier
>
> This is a bit of a catch-all ticket for general improvements to the Flink on 
> Docker experience.
> Things to improve:
>   - Make it possible to build a Docker image from your own flink-dist 
> directory as well as official releases.
>   - Make it possible to override the image name so a user can more easily 
> publish these images to their Docker repository
>   - Provide scripts that show how to properly run on Docker Swarm or similar 
> environments with overlay networking (Kubernetes) without using host 
> networking.
>   - Log to stdout rather than to files.
>   - Work properly with docker-compose for local deployment as well as 
> production deployments (Swarm/k8s)



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

[GitHub] flink pull request #3205: [FLINK-5635] Improve Docker tooling to make it eas...

[tonycox]

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

https://github.com/apache/flink/pull/3205#discussion_r100156082
  
--- Diff: flink-contrib/docker-flink/Dockerfile ---
@@ -36,22 +31,25 @@ ENV PATH $PATH:$FLINK_HOME/bin
 EXPOSE 8081
 EXPOSE 6123
 
+# flink-dist can point to a directory, a tarball on the local system, or a 
url to a tarball
+ARG flink_dist=NOT_SET
+
 # Install build dependencies and flink
+ADD $flink_dist $FLINK_INSTALL_PATH
 RUN set -x && \
   mkdir -p $FLINK_INSTALL_PATH && \
--- End diff --

and what if `$flink_dist` is url ? 
Should run `tar xzf` I think


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

[jira] [Created] (FLINK-5748) Make the ExecutionGraph's FutureExecutor a ScheduledExecutionService

[Stephan Ewen (JIRA)]

Stephan Ewen created FLINK-5748:
---

 Summary: Make the ExecutionGraph's FutureExecutor a 
ScheduledExecutionService
 Key: FLINK-5748
 URL: https://issues.apache.org/jira/browse/FLINK-5748
 Project: Flink
  Issue Type: Improvement
  Components: Distributed Coordination
Reporter: Stephan Ewen
Assignee: Stephan Ewen
 Fix For: 1.3.0


To handle timeouts and other scheduled actions more efficiently, the 
{{ExecutionGraph}} should use a {{ScheduledExecutorService}} for its futures 
and callbacks.



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

[jira] [Created] (FLINK-5747) Eager Scheduling should deploy all Tasks together

[Stephan Ewen (JIRA)]

Stephan Ewen created FLINK-5747:
---

 Summary: Eager Scheduling should deploy all Tasks together
 Key: FLINK-5747
 URL: https://issues.apache.org/jira/browse/FLINK-5747
 Project: Flink
  Issue Type: Bug
  Components: JobManager
Affects Versions: 1.2.0
Reporter: Stephan Ewen
Assignee: Stephan Ewen
 Fix For: 1.3.0


Currently, eager scheduling immediately triggers the scheduling for all 
vertices and their subtasks in topological order. 

This has two problems:

  - This works only, as long as resource acquisition is "synchronous". With 
dynamic resource acquisition in FLIP-6, the resources are returned as Futures 
which may complete out of order. This results in out-of-order (not in 
topological order) scheduling of tasks which does not work for streaming.

  - Deploying some tasks that depend on other tasks before it is clear that the 
other tasks have resources as well leads to situations where many 
deploy/recovery cycles happen before enough resources are available to get the 
job running fully.

For eager scheduling, we should allocate all resources in one chunk and then 
deploy once we know that all are available.

As a follow-up, the same should be done per pipelined component in lazy batch 
scheduling as well. That way we get lazy scheduling across blocking boundaries, 
and bulk (gang) scheduling in pipelined subgroups.

This also does not apply for efforts of fine grained recovery, where individual 
tasks request replacement resources.



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

[jira] [Commented] (FLINK-5738) Destroy created backend when task is canceled

[Stephan Ewen (JIRA)]

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

Stephan Ewen commented on FLINK-5738:
-

This is one more issue caused by the fact that the initialization of many 
fields is "lazy".
The best way to solve this is IMO to make the {{AbstractInvokable}} accept the 
{{Environment}} in the constructor and initialize all fields eagerly in the 
constructor.

> Destroy created backend when task is canceled
> -
>
> Key: FLINK-5738
> URL: https://issues.apache.org/jira/browse/FLINK-5738
> Project: Flink
>  Issue Type: Bug
>  Components: State Backends, Checkpointing
>Affects Versions: 1.2.0
>Reporter: Xiaogang Shi
>
> When a task is canceled, the {{ClosableRegistry}} will be closed in the 
> cancel thread. However, the task may still in the creation of 
> {{KeyedStateBackend}}, and it will fail to register the backend to the 
> {{ClosableRegistry}}. Because the backend is not assigned to the operator yet 
> (due to the exception), the backend will not be destroyed when the task 
> thread exits.
> A simple solution is to catch exception in the registering and destroy the 
> created backend in the case of failures. 



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

[jira] [Commented] (FLINK-5153) Allow setting custom application tags for Flink on YARN

[ASF GitHub Bot (JIRA)]

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

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

Github user patricklucas closed the pull request at:

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


> Allow setting custom application tags for Flink on YARN
> ---
>
> Key: FLINK-5153
> URL: https://issues.apache.org/jira/browse/FLINK-5153
> Project: Flink
>  Issue Type: Improvement
>  Components: YARN
>Reporter: Robert Metzger
>Assignee: Patrick Lucas
> Fix For: 1.3.0
>
>
> https://issues.apache.org/jira/browse/YARN-1399 added support in YARN to tag 
> applications.
> We should introduce a configuration variable in Flink allowing users to 
> specify a comma-separated list of tags they want to assign to their Flink on 
> YARN applications.



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

[GitHub] flink pull request #3268: [FLINK-5153] Support YARN application tags

[patricklucas]

Github user patricklucas closed the pull request at:

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


---
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-5635) Improve Docker tooling to make it easier to build images and launch Flink via Docker tools

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/3205#discussion_r100148697
  
--- Diff: flink-contrib/docker-flink/Dockerfile ---
@@ -36,22 +31,25 @@ ENV PATH $PATH:$FLINK_HOME/bin
 EXPOSE 8081
 EXPOSE 6123
 
+# flink-dist can point to a directory, a tarball on the local system, or a 
url to a tarball
+ARG flink_dist=NOT_SET
+
 # Install build dependencies and flink
+ADD $flink_dist $FLINK_INSTALL_PATH
 RUN set -x && \
   mkdir -p $FLINK_INSTALL_PATH && \
--- End diff --

place of error. when you add `$flink_dist` path of `$FLINK_INSTALL_PATH` 
will create. I propose to delete making dir here


> Improve Docker tooling to make it easier to build images and launch Flink via 
> Docker tools
> --
>
> Key: FLINK-5635
> URL: https://issues.apache.org/jira/browse/FLINK-5635
> Project: Flink
>  Issue Type: Improvement
>  Components: Docker
>Affects Versions: 1.2.0
>Reporter: Jamie Grier
>Assignee: Jamie Grier
>
> This is a bit of a catch-all ticket for general improvements to the Flink on 
> Docker experience.
> Things to improve:
>   - Make it possible to build a Docker image from your own flink-dist 
> directory as well as official releases.
>   - Make it possible to override the image name so a user can more easily 
> publish these images to their Docker repository
>   - Provide scripts that show how to properly run on Docker Swarm or similar 
> environments with overlay networking (Kubernetes) without using host 
> networking.
>   - Log to stdout rather than to files.
>   - Work properly with docker-compose for local deployment as well as 
> production deployments (Swarm/k8s)



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

[GitHub] flink pull request #3205: [FLINK-5635] Improve Docker tooling to make it eas...

[tonycox]

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

https://github.com/apache/flink/pull/3205#discussion_r100148697
  
--- Diff: flink-contrib/docker-flink/Dockerfile ---
@@ -36,22 +31,25 @@ ENV PATH $PATH:$FLINK_HOME/bin
 EXPOSE 8081
 EXPOSE 6123
 
+# flink-dist can point to a directory, a tarball on the local system, or a 
url to a tarball
+ARG flink_dist=NOT_SET
+
 # Install build dependencies and flink
+ADD $flink_dist $FLINK_INSTALL_PATH
 RUN set -x && \
   mkdir -p $FLINK_INSTALL_PATH && \
--- End diff --

place of error. when you add `$flink_dist` path of `$FLINK_INSTALL_PATH` 
will create. I propose to delete making dir here


---
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-5742) Breakpoints on documentation website

[Stephan Ewen (JIRA)]

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

Stephan Ewen commented on FLINK-5742:
-

/cc [~uce] [~alpinegizmo]

> Breakpoints on documentation website
> 
>
> Key: FLINK-5742
> URL: https://issues.apache.org/jira/browse/FLINK-5742
> Project: Flink
>  Issue Type: Improvement
>Affects Versions: 1.2.0
>Reporter: Colin Breame
>Priority: Trivial
>
> When reading the documentation website, I'm finding that unless I have the 
> browser window fully maximised, the content is formatted in the narrow layout.
> See https://ci.apache.org/projects/flink/flink-docs-release-1.2/
> *Expected behaviour:*
> The content should flow on the right-hand-side of the menu.
> *Actual behaviour*
> The menu is the full width of the window and the content flows below the menu.
> *Notes*
> Any width below 1200px causes the content to be placed below the full width 
> menu.  This 2/3 the width of my laptop screen.
> I would suggest making this smaller and consistent with the main project page 
> (http://flink.apache.org/) which has the breakpoint set to about 840px.



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

[jira] [Commented] (FLINK-5742) Breakpoints on documentation website

[Stephan Ewen (JIRA)]

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

Stephan Ewen commented on FLINK-5742:
-

+1, good idea

> Breakpoints on documentation website
> 
>
> Key: FLINK-5742
> URL: https://issues.apache.org/jira/browse/FLINK-5742
> Project: Flink
>  Issue Type: Improvement
>Affects Versions: 1.2.0
>Reporter: Colin Breame
>Priority: Trivial
>
> When reading the documentation website, I'm finding that unless I have the 
> browser window fully maximised, the content is formatted in the narrow layout.
> See https://ci.apache.org/projects/flink/flink-docs-release-1.2/
> *Expected behaviour:*
> The content should flow on the right-hand-side of the menu.
> *Actual behaviour*
> The menu is the full width of the window and the content flows below the menu.
> *Notes*
> Any width below 1200px causes the content to be placed below the full width 
> menu.  This 2/3 the width of my laptop screen.
> I would suggest making this smaller and consistent with the main project page 
> (http://flink.apache.org/) which has the breakpoint set to about 840px.



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

[jira] [Updated] (FLINK-5746) Add acceptEither and applyToEither to Flink's Future

[Stephan Ewen (JIRA)]

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

Stephan Ewen updated FLINK-5746:

Priority: Minor  (was: Major)

> Add acceptEither and applyToEither to Flink's Future
> 
>
> Key: FLINK-5746
> URL: https://issues.apache.org/jira/browse/FLINK-5746
> Project: Flink
>  Issue Type: Improvement
>  Components: Local Runtime
>Affects Versions: 1.3.0
>Reporter: Till Rohrmann
>Priority: Minor
>
> Flink's futures are missing the method {{acceptEither}} and {{applyToEither}} 
> in order to react to the completion of one of two futures. Adding them would 
> be helpful.



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

[jira] [Commented] (FLINK-5501) Determine whether the job starts from last JobManager failure

[Stephan Ewen (JIRA)]

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

Stephan Ewen commented on FLINK-5501:
-

I would actually go ahead with what you suggested originally and extend the 
{{RunningJobsRegistry}} to support the different states "CREATED", "RUNNING", 
"FINISHED".

I was only listing the other possible options for interested readers.

> Determine whether the job starts from last JobManager failure
> -
>
> Key: FLINK-5501
> URL: https://issues.apache.org/jira/browse/FLINK-5501
> Project: Flink
>  Issue Type: Sub-task
>  Components: JobManager
>Reporter: Zhijiang Wang
>Assignee: Zhijiang Wang
>
> When the {{JobManagerRunner}} grants leadership, it should check whether the 
> current job is already running or not. If the job is running, the 
> {{JobManager}} should reconcile itself (enter RECONCILING state) and waits 
> for the {{TaskManager}} reporting task status. Otherwise the {{JobManger}} 
> can schedule the {{ExecutionGraph}} in common way.
> The {{RunningJobsRegistry}} can provide the way to check the job running 
> status, but we should expand the current interface and fix the related 
> process to support this function.
> 1. {{RunningJobsRegistry}} sets RUNNING status after {{JobManagerRunner}} 
> granting leadership at the first time.
> 2. If the job finishes, the job status will be set FINISHED by 
> {{RunningJobsRegistry}} and the status will be deleted before exit. 
> 3. If the mini cluster starts multi {{JobManagerRunner}}, and the leader 
> {{JobManagerRunner}} already finishes the job to set the job status FINISHED, 
> other {{JobManagerRunner}} will exit after grants the leadership again.
> 4. If the {{JobManager}} fails, the job status will be still in RUNNING. So 
> if the {{JobManagerRunner}} (the previous or new one) grants leadership 
> again, it will check the job status and enters {{RECONCILING}} state.



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

[GitHub] flink issue #3268: [FLINK-5153] Support YARN application tags

[rmetzger]

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/3268
  
@patricklucas Can you manually close the PR? We usually do this 
automatically, but I forgot to add the right string the commit message.


---
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-5153) Allow setting custom application tags for Flink on YARN

[ASF GitHub Bot (JIRA)]

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

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

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/3268
  
@patricklucas Can you manually close the PR? We usually do this 
automatically, but I forgot to add the right string the commit message.


> Allow setting custom application tags for Flink on YARN
> ---
>
> Key: FLINK-5153
> URL: https://issues.apache.org/jira/browse/FLINK-5153
> Project: Flink
>  Issue Type: Improvement
>  Components: YARN
>Reporter: Robert Metzger
>Assignee: Patrick Lucas
> Fix For: 1.3.0
>
>
> https://issues.apache.org/jira/browse/YARN-1399 added support in YARN to tag 
> applications.
> We should introduce a configuration variable in Flink allowing users to 
> specify a comma-separated list of tags they want to assign to their Flink on 
> YARN applications.



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

[jira] [Commented] (FLINK-5718) Handle JVM Fatal Exceptions in Tasks

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/3276#discussion_r100128721
  
--- Diff: docs/setup/config.md ---
@@ -86,7 +86,7 @@ The default fraction for managed memory can be adjusted 
using the `taskmanager.m
 
 - `taskmanager.memory.segment-size`: The size of memory buffers used by 
the memory manager and the network stack in bytes (DEFAULT: 32768 (= 32 
KiBytes)).
 
-- `taskmanager.memory.preallocate`: Can be either of `true` or `false`. 
Specifies whether task managers should allocate all managed memory when 
starting up. (DEFAULT: false). When `taskmanager.memory.off-heap` is set to 
`true`, then it is advised that this configuration is also set to `true`.  If 
this configuration is set to `false` cleaning up of the allocated offheap 
memory happens only when the configured JVM parameter MaxDirectMemorySize is 
reached by triggering a full GC.
+- `taskmanager.memory.preallocate`: Can be either of `true` or `false`. 
Specifies whether task managers should allocate all managed memory when 
starting up. (DEFAULT: false). When `taskmanager.memory.off-heap` is set to 
`true`, then it is advised that this configuration is also set to `true`.  If 
this configuration is set to `false` cleaning up of the allocated offheap 
memory happens only when the configured JVM parameter MaxDirectMemorySize is 
reached by triggering a full GC. **Note:** For streaming setups, we highly 
recommend to set this value to `false` as the core state backends currently do 
not use the managed memory.
--- End diff --

That would probably be good


> Handle JVM Fatal Exceptions in Tasks
> 
>
> Key: FLINK-5718
> URL: https://issues.apache.org/jira/browse/FLINK-5718
> Project: Flink
>  Issue Type: Improvement
>  Components: Local Runtime
>Reporter: Stephan Ewen
>Assignee: Stephan Ewen
>
> The TaskManager catches and handles all types of exceptions right now (all 
> {{Throwables}}). The intention behind that is:
>   - Many {{Error}} subclasses are recoverable for the TaskManagers, such as 
> failure to load/link user code
>   - We want to give eager notifications to the JobManager in case something 
> in a task goes wrong.
> However, there are some exceptions which should probably simply terminate the 
> JVM, if caught in the task thread, because they may leave the JVM in a 
> dysfunctional limbo state:
>   - {{OutOfMemoryError}}
>   - {{InternalError}}
>   - {{UnknownError}}
>   - {{ZipError}}
> These are basically the subclasses of {{VirtualMachineError}}, except for 
> {{StackOverflowError}}, which is recoverable and usually recovered already by 
> the time the exception has been thrown and the stack unwound.



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

[GitHub] flink pull request #3276: [FLINK-5718] [core] TaskManagers exit the JVM on f...

[StephanEwen]

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

https://github.com/apache/flink/pull/3276#discussion_r100128721
  
--- Diff: docs/setup/config.md ---
@@ -86,7 +86,7 @@ The default fraction for managed memory can be adjusted 
using the `taskmanager.m
 
 - `taskmanager.memory.segment-size`: The size of memory buffers used by 
the memory manager and the network stack in bytes (DEFAULT: 32768 (= 32 
KiBytes)).
 
-- `taskmanager.memory.preallocate`: Can be either of `true` or `false`. 
Specifies whether task managers should allocate all managed memory when 
starting up. (DEFAULT: false). When `taskmanager.memory.off-heap` is set to 
`true`, then it is advised that this configuration is also set to `true`.  If 
this configuration is set to `false` cleaning up of the allocated offheap 
memory happens only when the configured JVM parameter MaxDirectMemorySize is 
reached by triggering a full GC.
+- `taskmanager.memory.preallocate`: Can be either of `true` or `false`. 
Specifies whether task managers should allocate all managed memory when 
starting up. (DEFAULT: false). When `taskmanager.memory.off-heap` is set to 
`true`, then it is advised that this configuration is also set to `true`.  If 
this configuration is set to `false` cleaning up of the allocated offheap 
memory happens only when the configured JVM parameter MaxDirectMemorySize is 
reached by triggering a full GC. **Note:** For streaming setups, we highly 
recommend to set this value to `false` as the core state backends currently do 
not use the managed memory.
--- End diff --

That would probably be good


---
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] [Resolved] (FLINK-5153) Allow setting custom application tags for Flink on YARN

[Robert Metzger (JIRA)]

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

Robert Metzger resolved FLINK-5153.
---
   Resolution: Fixed
Fix Version/s: 1.3.0

Thanks a lot for your contribution Patrick!

Change has been merged to 1.3: 
http://git-wip-us.apache.org/repos/asf/flink/commit/d9c116e5

> Allow setting custom application tags for Flink on YARN
> ---
>
> Key: FLINK-5153
> URL: https://issues.apache.org/jira/browse/FLINK-5153
> Project: Flink
>  Issue Type: Improvement
>  Components: YARN
>Reporter: Robert Metzger
>Assignee: Patrick Lucas
> Fix For: 1.3.0
>
>
> https://issues.apache.org/jira/browse/YARN-1399 added support in YARN to tag 
> applications.
> We should introduce a configuration variable in Flink allowing users to 
> specify a comma-separated list of tags they want to assign to their Flink on 
> YARN applications.



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

[jira] [Created] (FLINK-5746) Add acceptEither and applyToEither to Flink's Future

[Till Rohrmann (JIRA)]

Till Rohrmann created FLINK-5746:


 Summary: Add acceptEither and applyToEither to Flink's Future
 Key: FLINK-5746
 URL: https://issues.apache.org/jira/browse/FLINK-5746
 Project: Flink
  Issue Type: Improvement
  Components: Local Runtime
Affects Versions: 1.3.0
Reporter: Till Rohrmann


Flink's futures are missing the method {{acceptEither}} and {{applyToEither}} 
in order to react to the completion of one of two futures. Adding them would be 
helpful.



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

[jira] [Commented] (FLINK-5153) Allow setting custom application tags for Flink on YARN

[ASF GitHub Bot (JIRA)]

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

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

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/3268
  
I'll merge the change now.

I've also tested it.


> Allow setting custom application tags for Flink on YARN
> ---
>
> Key: FLINK-5153
> URL: https://issues.apache.org/jira/browse/FLINK-5153
> Project: Flink
>  Issue Type: Improvement
>  Components: YARN
>Reporter: Robert Metzger
>Assignee: Patrick Lucas
>
> https://issues.apache.org/jira/browse/YARN-1399 added support in YARN to tag 
> applications.
> We should introduce a configuration variable in Flink allowing users to 
> specify a comma-separated list of tags they want to assign to their Flink on 
> YARN applications.



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

[GitHub] flink issue #3268: [FLINK-5153] Support YARN application tags

[rmetzger]

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/3268
  
I'll merge the change now.

I've also tested 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-5701) FlinkKafkaProducer should check asyncException on checkpoints

[ASF GitHub Bot (JIRA)]

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

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

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/3278
  
I think we should allow users to disable the wait on flush, because it can 
substantially delay the confirmation of a checkpoint.
If a user favors fast checkpoints over complete data in Kafka (for example 
when a particular producer instance is used mostly for debugging purposes 
only), we should allow them to do that. The overhead for us making this 
configurable is very low, but the benefit for some users might be huge.


> FlinkKafkaProducer should check asyncException on checkpoints
> -
>
> Key: FLINK-5701
> URL: https://issues.apache.org/jira/browse/FLINK-5701
> Project: Flink
>  Issue Type: Bug
>  Components: Kafka Connector, Streaming Connectors
>Reporter: Tzu-Li (Gordon) Tai
>Priority: Critical
>
> Reported in ML: 
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Fink-KafkaProducer-Data-Loss-td11413.html
> The problem:
> The producer holds a {{pendingRecords}} value that is incremented on each 
> invoke() and decremented on each callback, used to check if the producer 
> needs to sync on pending callbacks on checkpoints.
> On each checkpoint, we should only consider the checkpoint succeeded iff 
> after flushing the {{pendingRecords == 0}} and {{asyncException == null}} 
> (currently, we’re only checking {{pendingRecords}}).
> A quick fix for this is to check and rethrow async exceptions in the 
> {{snapshotState}} method both before and after flushing and 
> {{pendingRecords}} becomes 0.



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

[GitHub] flink issue #3278: [FLINK-5701] [kafka] FlinkKafkaProducer should check asyn...

[rmetzger]

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/3278
  
I think we should allow users to disable the wait on flush, because it can 
substantially delay the confirmation of a checkpoint.
If a user favors fast checkpoints over complete data in Kafka (for example 
when a particular producer instance is used mostly for debugging purposes 
only), we should allow them to do that. The overhead for us making this 
configurable is very low, but the benefit for some users might be huge.


---
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-5353) Elasticsearch Sink loses well-formed documents when there are malformed documents

[ASF GitHub Bot (JIRA)]

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

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

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/3246
  
Thanks for the feedbacks. I'll rebase this PR soon.

@static-max thanks for the tip. I'll keep that in mind when updating the 
docs for this feature!


> Elasticsearch Sink loses well-formed documents when there are malformed 
> documents
> -
>
> Key: FLINK-5353
> URL: https://issues.apache.org/jira/browse/FLINK-5353
> Project: Flink
>  Issue Type: Bug
>  Components: Streaming Connectors
>Affects Versions: 1.1.3
>Reporter: Flavio Pompermaier
>Assignee: Tzu-Li (Gordon) Tai
>




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

[GitHub] flink issue #3246: [FLINK-5353] [elasticsearch] User-provided failure handle...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/3246
  
Thanks for the feedbacks. I'll rebase this PR soon.

@static-max thanks for the tip. I'll keep that in mind when updating the 
docs for this feature!


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

[jira] [Created] (FLINK-5745) Set uncaught exception handler for Netty threads

[Ufuk Celebi (JIRA)]

Ufuk Celebi created FLINK-5745:
--

 Summary: Set uncaught exception handler for Netty threads
 Key: FLINK-5745
 URL: https://issues.apache.org/jira/browse/FLINK-5745
 Project: Flink
  Issue Type: Improvement
  Components: Network
Reporter: Ufuk Celebi
Priority: Minor


We pass a thread factory for the Netty event loop threads (see {{NettyServer}} 
and {{NettyClient}}), but don't set an uncaught exception handler. Let's add a 
JVM terminating handler that exits the process in cause of fatal errors.






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

[jira] [Updated] (FLINK-5744) Check remote connection in Flink-shell

[Anton Solovev (JIRA)]

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

Anton Solovev updated FLINK-5744:
-
Summary: Check remote connection in Flink-shell  (was: Checking remote 
connection in Flink-shell)

> Check remote connection in Flink-shell
> --
>
> Key: FLINK-5744
> URL: https://issues.apache.org/jira/browse/FLINK-5744
> Project: Flink
>  Issue Type: Improvement
>  Components: Scala Shell
>Reporter: Anton Solovev
>Priority: Minor
>




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

[jira] [Commented] (FLINK-4540) Detached job execution may prevent cluster shutdown

[ASF GitHub Bot (JIRA)]

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

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

GitHub user MayerRoman opened a pull request:

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

[FLINK-4540][yarn] Delayed cluster startup until the job has been fully 
assembled, in the case of launching detached jobs

[FLINK-4540] Detached job execution may prevent cluster shutdown.

**workflow description:**
In the process of launching detached jobs on YARN 
`FlikYarnSession#createCluster` method call 
`AbstractYarnClusterDescriptor#deploy`, this method triggers deployment of a 
flink-cluster on YARN.
After that `CliFrontend#executeProgram` calls `ClusterClient#run`, causing 
the start of the job preparation process.
Further method `YarnClusterClient#submitJob` calls 
`YarnClusterClient#stopAfterJob` which ensures that the YarnJobManager shuts 
down after the job completes, and calls `ClusterClient#runDetached` that sends 
job on the cluster.

**how the bug occurs:**
If the error will be thrown after `AbstractYarnClusterDescriptor#deploy`, 
but before sending job on the cluster, the jobmanager never shuts down the 
cluster.

**changes:**
This pull request defers cluster startup until the job has been fully 
assembled.

**additional Information:**
- these changes do not affect the work of FLIP-6

- explanation for rows 330-331 in `FlinkYarnSessionCli`:

After` "flink run ..."` in command line Flink creates instance of 
`CliFronted`.
Befor creating instance of `CliFrontend` static block in this class is 
executed. It creates instances of `FlinkYarnSessionCli`, `FlinkYarnCli`, 
`DefaultCli `and puts it in static LinkedList.

In the case of real work, every time a new JVM starts.

In the case of executing tests in flink-yarn-tests, launchig job in 
different modes occures in one JVM and static block in `CliFrontend ` is 
executed only once.
Therefore once created instance of `FlinkYarnSessionCli ` is reused by all 
tests in class.

`YARNSessionCapacitySchedulerITCase#testDetachedPerJobYarnCluster` and 
`#testDetachedPerJobYarnClusterWithStreamingJob` during execution change 
`private boolean detachedMode` in `FlinkYarnSessionCli `to true.

If after them will run 
`YARNSessionCapacitySchedulerITCase#perJobYarnCluster` or 
`#perJobYarnClusterWithParallelism`, because of the changed `detachedMode `in 
`FlinkYarnSessionCli `they will go the wrong execution way.

A previous version of this part of the code was based on the fact that the 
field is a priori false and only checks whether it is necessary to change it to 
true.
```
if (cmd.hasOption(DETACHED.getOpt()) || 
cmd.hasOption(CliFrontendParser.DETACHED_OPTION.getOpt())) {
this.detachedMode = true;
yarnClusterDescriptor.setDetachedMode(true);
}
```
The new version of this part of the code changes `detachedMode `anyway, 
whereby if it was true and it isn't launch of detached Job, it will be changed 
into false.

```
this.detachedMode = cmd.hasOption(DETACHED.getOpt()) || 
cmd.hasOption(CliFrontendParser.DETACHED_OPTION.getOpt());
yarnClusterDescriptor.setDetachedMode(this.detachedMode);
```

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

$ git pull https://github.com/MayerRoman/flink FLINK-4540

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

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


commit cca5c31767a76a560e66193f28e023210d592dbf
Author: Roman Maier 
Date:   2017-02-08T08:04:15Z

[FLINK-4540][yarn] Delayed cluster startup until the job has been fully 
assembled, in the case of launching detached jobs




> Detached job execution may prevent cluster shutdown
> ---
>
> Key: FLINK-4540
> URL: https://issues.apache.org/jira/browse/FLINK-4540
> Project: Flink
>  Issue Type: Bug
>  Components: YARN
>Affects Versions: 1.2.0, 1.1.2
>Reporter: Maximilian Michels
>Assignee: Roman Maier
>Priority: Minor
>
> There is a problem with the detached execution of jobs. This can prevent 
> cluster shutdown 1) when eager jobs are executed, i.e. the job calls 
> `collect()/count()`, and 2) when the user jar doesn't contain a job. 
> 1) For example, {{./flink -d -m yarn-cluster -yn 1 
> ../examples/batch/WordCount.jar}} will throw an exception and only disconnect 
> the YarnClusterClient afterwards. In detached mode, the code assumes the 
> cluster is 

[GitHub] flink pull request #3287: [FLINK-4540][yarn] Delayed cluster startup until t...

[MayerRoman]

GitHub user MayerRoman opened a pull request:

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

[FLINK-4540][yarn] Delayed cluster startup until the job has been fully 
assembled, in the case of launching detached jobs

[FLINK-4540] Detached job execution may prevent cluster shutdown.

**workflow description:**
In the process of launching detached jobs on YARN 
`FlikYarnSession#createCluster` method call 
`AbstractYarnClusterDescriptor#deploy`, this method triggers deployment of a 
flink-cluster on YARN.
After that `CliFrontend#executeProgram` calls `ClusterClient#run`, causing 
the start of the job preparation process.
Further method `YarnClusterClient#submitJob` calls 
`YarnClusterClient#stopAfterJob` which ensures that the YarnJobManager shuts 
down after the job completes, and calls `ClusterClient#runDetached` that sends 
job on the cluster.

**how the bug occurs:**
If the error will be thrown after `AbstractYarnClusterDescriptor#deploy`, 
but before sending job on the cluster, the jobmanager never shuts down the 
cluster.

**changes:**
This pull request defers cluster startup until the job has been fully 
assembled.

**additional Information:**
- these changes do not affect the work of FLIP-6

- explanation for rows 330-331 in `FlinkYarnSessionCli`:

After` "flink run ..."` in command line Flink creates instance of 
`CliFronted`.
Befor creating instance of `CliFrontend` static block in this class is 
executed. It creates instances of `FlinkYarnSessionCli`, `FlinkYarnCli`, 
`DefaultCli `and puts it in static LinkedList.

In the case of real work, every time a new JVM starts.

In the case of executing tests in flink-yarn-tests, launchig job in 
different modes occures in one JVM and static block in `CliFrontend ` is 
executed only once.
Therefore once created instance of `FlinkYarnSessionCli ` is reused by all 
tests in class.

`YARNSessionCapacitySchedulerITCase#testDetachedPerJobYarnCluster` and 
`#testDetachedPerJobYarnClusterWithStreamingJob` during execution change 
`private boolean detachedMode` in `FlinkYarnSessionCli `to true.

If after them will run 
`YARNSessionCapacitySchedulerITCase#perJobYarnCluster` or 
`#perJobYarnClusterWithParallelism`, because of the changed `detachedMode `in 
`FlinkYarnSessionCli `they will go the wrong execution way.

A previous version of this part of the code was based on the fact that the 
field is a priori false and only checks whether it is necessary to change it to 
true.
```
if (cmd.hasOption(DETACHED.getOpt()) || 
cmd.hasOption(CliFrontendParser.DETACHED_OPTION.getOpt())) {
this.detachedMode = true;
yarnClusterDescriptor.setDetachedMode(true);
}
```
The new version of this part of the code changes `detachedMode `anyway, 
whereby if it was true and it isn't launch of detached Job, it will be changed 
into false.

```
this.detachedMode = cmd.hasOption(DETACHED.getOpt()) || 
cmd.hasOption(CliFrontendParser.DETACHED_OPTION.getOpt());
yarnClusterDescriptor.setDetachedMode(this.detachedMode);
```

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

$ git pull https://github.com/MayerRoman/flink FLINK-4540

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

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


commit cca5c31767a76a560e66193f28e023210d592dbf
Author: Roman Maier 
Date:   2017-02-08T08:04:15Z

[FLINK-4540][yarn] Delayed cluster startup until the job has been fully 
assembled, in the case of launching detached jobs




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

[jira] [Created] (FLINK-5744) Checking remote connection in Flink-shell

[Anton Solovev (JIRA)]

Anton Solovev created FLINK-5744:


 Summary: Checking remote connection in Flink-shell
 Key: FLINK-5744
 URL: https://issues.apache.org/jira/browse/FLINK-5744
 Project: Flink
  Issue Type: Improvement
  Components: Scala Shell
Reporter: Anton Solovev
Priority: Minor






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

[GitHub] flink issue #3246: [FLINK-5353] [elasticsearch] User-provided failure handle...

[static-max]

Github user static-max commented on the issue:

https://github.com/apache/flink/pull/3246
  
Looks great!
One note: In your example ExampleActionRequestFailureHandler you have to 
unwrap the Exception, as it is typical looks like:

`RemoteTransportException[[Richard 
Rider][127.0.0.1:9301][indices:data/write/bulk[s]]]; nested: 
RemoteTransportException[[Richard 
Rider][127.0.0.1:9301][indices:data/write/bulk[s][p]]]; nested: 
EsRejectedExecutionException[rejected execution of 
org.elasticsearch.transport.TransportService$4@e5c47a1 on 
EsThreadPoolExecutor[bulk, queue capacity = 1, 
org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor@7e2d5cc5[Running, 
pool size = 8, active threads = 8, queued tasks = 1, completed tasks = 119]]];`

In my implementation I use Apache commons:
`
ExceptionUtils.indexOfThrowable(throwable, 
EsRejectedExecutionException.class) >= 0
`


---
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-5353) Elasticsearch Sink loses well-formed documents when there are malformed documents

[ASF GitHub Bot (JIRA)]

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

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

Github user static-max commented on the issue:

https://github.com/apache/flink/pull/3246
  
Looks great!
One note: In your example ExampleActionRequestFailureHandler you have to 
unwrap the Exception, as it is typical looks like:

`RemoteTransportException[[Richard 
Rider][127.0.0.1:9301][indices:data/write/bulk[s]]]; nested: 
RemoteTransportException[[Richard 
Rider][127.0.0.1:9301][indices:data/write/bulk[s][p]]]; nested: 
EsRejectedExecutionException[rejected execution of 
org.elasticsearch.transport.TransportService$4@e5c47a1 on 
EsThreadPoolExecutor[bulk, queue capacity = 1, 
org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor@7e2d5cc5[Running, 
pool size = 8, active threads = 8, queued tasks = 1, completed tasks = 119]]];`

In my implementation I use Apache commons:
`
ExceptionUtils.indexOfThrowable(throwable, 
EsRejectedExecutionException.class) >= 0
`


> Elasticsearch Sink loses well-formed documents when there are malformed 
> documents
> -
>
> Key: FLINK-5353
> URL: https://issues.apache.org/jira/browse/FLINK-5353
> Project: Flink
>  Issue Type: Bug
>  Components: Streaming Connectors
>Affects Versions: 1.1.3
>Reporter: Flavio Pompermaier
>Assignee: Tzu-Li (Gordon) Tai
>




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

[jira] [Assigned] (FLINK-5476) Fail fast if trying to submit a job to a non-existing Flink cluster

[Dmitrii Kniazev (JIRA)]

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

Dmitrii Kniazev reassigned FLINK-5476:
--

Assignee: Dmitrii Kniazev

> Fail fast if trying to submit a job to a non-existing Flink cluster
> ---
>
> Key: FLINK-5476
> URL: https://issues.apache.org/jira/browse/FLINK-5476
> Project: Flink
>  Issue Type: Improvement
>  Components: Client
>Affects Versions: 1.2.0, 1.3.0
>Reporter: Till Rohrmann
>Assignee: Dmitrii Kniazev
>Priority: Minor
>
> In case of entering the wrong job manager address when submitting a job via 
> {{flink run}}, the {{JobClientActor}} waits per default {{60 s}} until a 
> {{JobClientActorConnectionException}}, indicating that the {{JobManager}} is 
> no longer reachable, is thrown. In order to fail fast in case of wrong 
> connection information, we could change it such that it uses initially a much 
> lower timeout and only increases the timeout if it had at least once 
> successfully connected to a {{JobManager}} before.



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

[jira] [Comment Edited] (FLINK-5728) FlinkKafkaProducer should flush on checkpoint by default

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

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

Tzu-Li (Gordon) Tai edited comment on FLINK-5728 at 2/8/17 11:20 AM:
-

Related:

I've recently gathered thoughts from [~uce] and [~till.rohrmann] that the 
methods {{setLogFailuresOnly}} and {{setFlushOnCheckpoint}} are 
over-complicating at-least-once guarantees for the Kafka producer. You can see 
by the docs for it here: 
https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/connectors/kafka.html#kafka-producers-and-fault-tolerance,
 rather complicated for the user ;-)

I think there was 2 main reasons why the {{setFlushOnCheckpoint}} was added in 
the first place (see https://github.com/apache/flink/pull/2108, I'm basically 
just wrapping up):
- originally there was no flushing, so to avoid breaking user code behaviour, 
the method was added to allow users who know they need at-least-once to turn it 
on.
- make sure the producer was flexible and works in different situations.

There's 2 ways to go from here:
1. Once we make flush on checkpoint enabled by default, the user won't have to 
do anything for at-least-once as long as they turn on checkpointing. Still 
retain the methods for "advanced users" who need to tweak the settings.
2. Apart from changing default, also remove the methods completely. You'll 
always have at-least-once if checkpointing is enabled, at-most-once otherwise.

I personally prefer option 2, as I don't see the need to turn flushing off, 
when you somehow have needs for guarantees and turned on checkpointing in the 
first place. Also, the API is just simpler for the user and less things to 
worry about. I may be missing something though.

cc [~rmetzger] I'd also like to hear your take on this!


was (Author: tzulitai):
Related:

I've recently gathered thoughts from [~uce] and [~till.rohrmann] that the 
methods `setLogFailuresOnly` and `setFlushOnCheckpoint` are over-complicating 
at-least-once guarantees for the Kafka producer. You can see by the docs for it 
here: 
https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/connectors/kafka.html#kafka-producers-and-fault-tolerance,
 rather complicated for the user ;-)

I think there was 2 main reasons why the `setFlushOnCheckpoint` was added in 
the first place (see https://github.com/apache/flink/pull/2108, I'm basically 
wrapping up):
- originally there was no flushing, so to avoid breaking user code behaviour, 
the method was added to allow users who know they need at-least-once to turn it 
on.
- make sure the producer was flexible and works in different situations.

There's 2 ways to go from here:
1. Once we make flush on checkpoint enabled by default, the user won't have to 
do anything for at-least-once as long as they turn on checkpointing. Still 
retain the methods for "advanced users" who need to tweak the settings.
2. Apart from changing default, also remove the methods completely. You'll 
always have at-least-once if checkpointing is enabled, at-most-once otherwise.

I personally prefer option 2, as I don't see the need to turn flushing off, 
when you somehow have needs for guarantees and turned on checkpointing in the 
first place. Also, the API is just simpler for the user and less things to 
worry about. I may be missing something though.

cc [~rmetzger] I'd also like to hear your take on this!

> FlinkKafkaProducer should flush on checkpoint by default
> 
>
> Key: FLINK-5728
> URL: https://issues.apache.org/jira/browse/FLINK-5728
> Project: Flink
>  Issue Type: Improvement
>  Components: Kafka Connector
>Reporter: Tzu-Li (Gordon) Tai
>
> As discussed in FLINK-5702, it might be a good idea to let the 
> FlinkKafkaProducer flush on checkpoints by default. Currently, it is disabled 
> by default.
> It's a very simple change, but we should think about whether or not we want 
> to break user behaviour, or have proper usage migration.



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

[jira] [Comment Edited] (FLINK-5728) FlinkKafkaProducer should flush on checkpoint by default

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

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

Tzu-Li (Gordon) Tai edited comment on FLINK-5728 at 2/8/17 11:20 AM:
-

Related:

I've recently gathered thoughts from [~uce] and [~till.rohrmann] that the 
methods {{setLogFailuresOnly}} and {{setFlushOnCheckpoint}} are 
over-complicating at-least-once guarantees for the Kafka producer. You can see 
by the docs for it here: 
https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/connectors/kafka.html#kafka-producers-and-fault-tolerance,
 rather complicated for the user ;-)

I think there were 2 main reasons why the {{setFlushOnCheckpoint}} was added in 
the first place (see https://github.com/apache/flink/pull/2108, I'm basically 
just wrapping up):
- originally there was no flushing, so to avoid breaking user code behaviour, 
the method was added to allow users who know they need at-least-once to turn it 
on.
- make sure the producer was flexible and works in different situations.

There's 2 ways to go from here:
1. Once we make flush on checkpoint enabled by default, the user won't have to 
do anything for at-least-once as long as they turn on checkpointing. Still 
retain the methods for "advanced users" who need to tweak the settings.
2. Apart from changing default, also remove the methods completely. You'll 
always have at-least-once if checkpointing is enabled, at-most-once otherwise.

I personally prefer option 2, as I don't see the need to turn flushing off, 
when you somehow have needs for guarantees and turned on checkpointing in the 
first place. Also, the API is just simpler for the user and less things to 
worry about. I may be missing something though.

cc [~rmetzger] I'd also like to hear your take on this!


was (Author: tzulitai):
Related:

I've recently gathered thoughts from [~uce] and [~till.rohrmann] that the 
methods {{setLogFailuresOnly}} and {{setFlushOnCheckpoint}} are 
over-complicating at-least-once guarantees for the Kafka producer. You can see 
by the docs for it here: 
https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/connectors/kafka.html#kafka-producers-and-fault-tolerance,
 rather complicated for the user ;-)

I think there was 2 main reasons why the {{setFlushOnCheckpoint}} was added in 
the first place (see https://github.com/apache/flink/pull/2108, I'm basically 
just wrapping up):
- originally there was no flushing, so to avoid breaking user code behaviour, 
the method was added to allow users who know they need at-least-once to turn it 
on.
- make sure the producer was flexible and works in different situations.

There's 2 ways to go from here:
1. Once we make flush on checkpoint enabled by default, the user won't have to 
do anything for at-least-once as long as they turn on checkpointing. Still 
retain the methods for "advanced users" who need to tweak the settings.
2. Apart from changing default, also remove the methods completely. You'll 
always have at-least-once if checkpointing is enabled, at-most-once otherwise.

I personally prefer option 2, as I don't see the need to turn flushing off, 
when you somehow have needs for guarantees and turned on checkpointing in the 
first place. Also, the API is just simpler for the user and less things to 
worry about. I may be missing something though.

cc [~rmetzger] I'd also like to hear your take on this!

> FlinkKafkaProducer should flush on checkpoint by default
> 
>
> Key: FLINK-5728
> URL: https://issues.apache.org/jira/browse/FLINK-5728
> Project: Flink
>  Issue Type: Improvement
>  Components: Kafka Connector
>Reporter: Tzu-Li (Gordon) Tai
>
> As discussed in FLINK-5702, it might be a good idea to let the 
> FlinkKafkaProducer flush on checkpoints by default. Currently, it is disabled 
> by default.
> It's a very simple change, but we should think about whether or not we want 
> to break user behaviour, or have proper usage migration.



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

[jira] [Commented] (FLINK-5728) FlinkKafkaProducer should flush on checkpoint by default

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

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

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


Related:

I've recently gathered thoughts from [~uce] and [~till.rohrmann] that the 
methods `setLogFailuresOnly` and `setFlushOnCheckpoint` are over-complicating 
at-least-once guarantees for the Kafka producer. You can see by the docs for it 
here: 
https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/connectors/kafka.html#kafka-producers-and-fault-tolerance,
 rather complicated for the user ;-)

I think there was 2 main reasons why the `setFlushOnCheckpoint` was added in 
the first place (see https://github.com/apache/flink/pull/2108, I'm basically 
wrapping up):
- originally there was no flushing, so to avoid breaking user code behaviour, 
the method was added to allow users who know they need at-least-once to turn it 
on.
- make sure the producer was flexible and works in different situations.

There's 2 ways to go from here:
1. Once we make flush on checkpoint enabled by default, the user won't have to 
do anything for at-least-once as long as they turn on checkpointing. Still 
retain the methods for "advanced users" who need to tweak the settings.
2. Apart from changing default, also remove the methods completely. You'll 
always have at-least-once if checkpointing is enabled, at-most-once otherwise.

I personally prefer option 2, as I don't see the need to turn flushing off, 
when you somehow have needs for guarantees and turned on checkpointing in the 
first place. Also, the API is just simpler for the user and less things to 
worry about. I may be missing something though.

cc [~rmetzger] I'd also like to hear your take on this!

> FlinkKafkaProducer should flush on checkpoint by default
> 
>
> Key: FLINK-5728
> URL: https://issues.apache.org/jira/browse/FLINK-5728
> Project: Flink
>  Issue Type: Improvement
>  Components: Kafka Connector
>Reporter: Tzu-Li (Gordon) Tai
>
> As discussed in FLINK-5702, it might be a good idea to let the 
> FlinkKafkaProducer flush on checkpoints by default. Currently, it is disabled 
> by default.
> It's a very simple change, but we should think about whether or not we want 
> to break user behaviour, or have proper usage migration.



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

[jira] [Commented] (FLINK-5725) Support JOIN between two streams in the SQL API

[radu (JIRA)]

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

radu commented on FLINK-5725:
-

Ok - thanks. 
I will work over the next weeks to come up with a design and a proposal for 
this to receive feedback for it and then we can start the development

> Support JOIN between two streams in the SQL API
> ---
>
> Key: FLINK-5725
> URL: https://issues.apache.org/jira/browse/FLINK-5725
> Project: Flink
>  Issue Type: New Feature
>  Components: Table API & SQL
>Reporter: Haohui Mai
>Assignee: radu
>
> As described in the title.
> This jira proposes to support joining two streaming tables in the SQL API.



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

[jira] [Commented] (FLINK-5701) FlinkKafkaProducer should check asyncException on checkpoints

[ASF GitHub Bot (JIRA)]

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

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

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/3278
  
Thanks for the detailed review @tillrohrmann, I'll follow-up and address 
your comments.

Regarding removing the `setFlushOnCheckpoint`:
I think it was added at first to provide flexibility for users who know 
what they are doing, and making sure that the producer will be able to work in 
all environments (see comments in #2108).

However, recently I've also gathered opinions (from you and others) about 
the settings over complicating at-least-once guarantees for the producer, and I 
have the feeling we can remove it starting from the next release.

There is FLINK-5728 to enable flushing by default (currently the default is 
no flushing). I'll incorporate your opinion on this to that JIRA, and decide 
there if we only want to disable if by default or remove it completely.



> FlinkKafkaProducer should check asyncException on checkpoints
> -
>
> Key: FLINK-5701
> URL: https://issues.apache.org/jira/browse/FLINK-5701
> Project: Flink
>  Issue Type: Bug
>  Components: Kafka Connector, Streaming Connectors
>Reporter: Tzu-Li (Gordon) Tai
>Priority: Critical
>
> Reported in ML: 
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Fink-KafkaProducer-Data-Loss-td11413.html
> The problem:
> The producer holds a {{pendingRecords}} value that is incremented on each 
> invoke() and decremented on each callback, used to check if the producer 
> needs to sync on pending callbacks on checkpoints.
> On each checkpoint, we should only consider the checkpoint succeeded iff 
> after flushing the {{pendingRecords == 0}} and {{asyncException == null}} 
> (currently, we’re only checking {{pendingRecords}}).
> A quick fix for this is to check and rethrow async exceptions in the 
> {{snapshotState}} method both before and after flushing and 
> {{pendingRecords}} becomes 0.



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

[GitHub] flink issue #3278: [FLINK-5701] [kafka] FlinkKafkaProducer should check asyn...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/3278
  
Thanks for the detailed review @tillrohrmann, I'll follow-up and address 
your comments.

Regarding removing the `setFlushOnCheckpoint`:
I think it was added at first to provide flexibility for users who know 
what they are doing, and making sure that the producer will be able to work in 
all environments (see comments in #2108).

However, recently I've also gathered opinions (from you and others) about 
the settings over complicating at-least-once guarantees for the producer, and I 
have the feeling we can remove it starting from the next release.

There is FLINK-5728 to enable flushing by default (currently the default is 
no flushing). I'll incorporate your opinion on this to that JIRA, and decide 
there if we only want to disable if by default or 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-1579) Create a Flink History Server

[ASF GitHub Bot (JIRA)]

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

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

GitHub user zentol opened a pull request:

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

[FLINK-1579] [WIP] Implement Standalone HistoryServer

This PR is a work-in-progress view over a standalone History Server (HS).

JobManagers may send completed jobs to the HistoryServer for them to be 
archived. Upon receiving an ArchivedExecutionGraph the HS pre-computes all 
possible REST requests and writes them into files. The files are arranged in a 
directory structure corresponding to the REST API.

The HS can be started by calling `./bin/historyserver.sh start`, similar to 
the JM/TM. Various config options exist for the HS that mostly mirror the 
web-ui/RPC options of the JM.

The HS uses a slightly modified web-ui; basically it only shows the 
"Completed Jobs" page. To not duplicate the everything I've added 2 files, 
`index2.jade` and `index2.coffee`, to the build script. The resulting 
`index2.html` file will be loaded when the browser requests the `index.html`.

In order to re-use the JSON generation code that previously was contained 
in various handlers a giant utility `JsonUtils` class was created. This class 
now contains a variety of static methods that generate the JSON responses. As a 
result most handlers were reduced to one-liners, bar some sanity-checks.

In regard to tests we verify that the HS creates all expected files upon 
receiving an ExecutionGraph.
Furthermore, the newly created JsonUtils are mostly tested (the new 
checkpoint stats aren't tested); so we have tests for the REST responses now, 
which is neat.

I'm not opening a proper PR yet as i have to go through all changes once 
again in detail, but it works (locally and on a cluster) so i wanted people to 
try it out and get some feedback.

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

$ git pull https://github.com/zentol/flink 1579_history_server_b

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

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


commit 0fdfeec0c86cba60d271d38cfbce7e4ae759b700
Author: zentol 
Date:   2016-10-17T10:55:19Z

Add AccessExecutionVertex#getPriorExecutions()

commit 18c4cc6a9e8f3c9b772bcfe8f866e07d2f7304ce
Author: zentol 
Date:   2017-01-30T15:06:13Z

[FLINK-5645] EG stuff

commit fcc4def5251086d4e37901c58bc47785e1d90788
Author: zentol 
Date:   2017-01-24T09:13:24Z

[FLINK-1579] Implement History Server - Frontend

commit 2cc6b736c0c5c78903b85f9c1a9ccde8c3ee70b8
Author: zentol 
Date:   2016-10-21T12:29:30Z

[FLINK-1579] Implement History Server - Backend

commit 0047ae53b9f2f79eee9ec7e76195559b32dbeb20
Author: zentol 
Date:   2017-02-08T08:58:01Z

[FLINK-1579] Implement History Server - Backend - Tests

commit 730548a7d88c56a2cde235e3d7d92dbf676611b7
Author: zentol 
Date:   2017-02-08T08:58:22Z

Use JsonUtils in handlers

commit adcc161e46f817e80301d1fb885cdef4a8679a71
Author: zentol 
Date:   2017-02-08T10:23:56Z

Rebuild web-frontend

commit 3227fc2a12e8aeaaf111339833123da708ccea70
Author: zentol 
Date:   2017-02-08T10:24:14Z

tmp streaming example with checkpointing




> Create a Flink History Server
> -
>
> Key: FLINK-1579
> URL: https://issues.apache.org/jira/browse/FLINK-1579
> Project: Flink
>  Issue Type: New Feature
>  Components: Distributed Coordination
>Affects Versions: 0.9
>Reporter: Robert Metzger
>Assignee: Chesnay Schepler
>
> Right now its not possible to analyze the job results for jobs that ran on 
> YARN, because we'll loose the information once the JobManager has stopped.
> Therefore, I propose to implement a "Flink History Server" which serves  the 
> results from these jobs.
> I haven't started thinking about the implementation, but I suspect it 
> involves some JSON files stored in HDFS :)



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

[GitHub] flink pull request #3286: [FLINK-1579] [WIP] Implement Standalone HistorySer...

[zentol]

GitHub user zentol opened a pull request:

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

[FLINK-1579] [WIP] Implement Standalone HistoryServer

This PR is a work-in-progress view over a standalone History Server (HS).

JobManagers may send completed jobs to the HistoryServer for them to be 
archived. Upon receiving an ArchivedExecutionGraph the HS pre-computes all 
possible REST requests and writes them into files. The files are arranged in a 
directory structure corresponding to the REST API.

The HS can be started by calling `./bin/historyserver.sh start`, similar to 
the JM/TM. Various config options exist for the HS that mostly mirror the 
web-ui/RPC options of the JM.

The HS uses a slightly modified web-ui; basically it only shows the 
"Completed Jobs" page. To not duplicate the everything I've added 2 files, 
`index2.jade` and `index2.coffee`, to the build script. The resulting 
`index2.html` file will be loaded when the browser requests the `index.html`.

In order to re-use the JSON generation code that previously was contained 
in various handlers a giant utility `JsonUtils` class was created. This class 
now contains a variety of static methods that generate the JSON responses. As a 
result most handlers were reduced to one-liners, bar some sanity-checks.

In regard to tests we verify that the HS creates all expected files upon 
receiving an ExecutionGraph.
Furthermore, the newly created JsonUtils are mostly tested (the new 
checkpoint stats aren't tested); so we have tests for the REST responses now, 
which is neat.

I'm not opening a proper PR yet as i have to go through all changes once 
again in detail, but it works (locally and on a cluster) so i wanted people to 
try it out and get some feedback.

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

$ git pull https://github.com/zentol/flink 1579_history_server_b

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

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


commit 0fdfeec0c86cba60d271d38cfbce7e4ae759b700
Author: zentol 
Date:   2016-10-17T10:55:19Z

Add AccessExecutionVertex#getPriorExecutions()

commit 18c4cc6a9e8f3c9b772bcfe8f866e07d2f7304ce
Author: zentol 
Date:   2017-01-30T15:06:13Z

[FLINK-5645] EG stuff

commit fcc4def5251086d4e37901c58bc47785e1d90788
Author: zentol 
Date:   2017-01-24T09:13:24Z

[FLINK-1579] Implement History Server - Frontend

commit 2cc6b736c0c5c78903b85f9c1a9ccde8c3ee70b8
Author: zentol 
Date:   2016-10-21T12:29:30Z

[FLINK-1579] Implement History Server - Backend

commit 0047ae53b9f2f79eee9ec7e76195559b32dbeb20
Author: zentol 
Date:   2017-02-08T08:58:01Z

[FLINK-1579] Implement History Server - Backend - Tests

commit 730548a7d88c56a2cde235e3d7d92dbf676611b7
Author: zentol 
Date:   2017-02-08T08:58:22Z

Use JsonUtils in handlers

commit adcc161e46f817e80301d1fb885cdef4a8679a71
Author: zentol 
Date:   2017-02-08T10:23:56Z

Rebuild web-frontend

commit 3227fc2a12e8aeaaf111339833123da708ccea70
Author: zentol 
Date:   2017-02-08T10:24:14Z

tmp streaming example with checkpointing




---
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-5701) FlinkKafkaProducer should check asyncException on checkpoints

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/3278#discussion_r100036593
  
--- Diff: 
flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBaseTest.java
 ---
@@ -293,6 +293,61 @@ public void run() {
testHarness.close();
}
 
+   /**
+* This test is meant to assure that testAtLeastOnceProducer is valid 
by testing that if flushing is disabled,
+* the snapshot method does indeed finishes without waiting for pending 
records;
+* we set a timeout because the test will not finish if the logic is 
broken
+*/
+   @Test(timeout=5000)
+   public void testDoesNotWaitForPendingRecordsIfFlushingDisabled() throws 
Throwable {
+   final OneShotLatch inputLatch = new OneShotLatch();
+
+   final DummyFlinkKafkaProducer producer = new 
DummyFlinkKafkaProducer<>(
--- End diff --

That makes sense. I was just reusing the {{MockProducer}} implementation in 
this class, which is needed in other tests to be able to provide success / 
failure fake callback completions. In this test we don't need to do that, so a 
simple {{mock(KafkaProducer.class)}} should do.


> FlinkKafkaProducer should check asyncException on checkpoints
> -
>
> Key: FLINK-5701
> URL: https://issues.apache.org/jira/browse/FLINK-5701
> Project: Flink
>  Issue Type: Bug
>  Components: Kafka Connector, Streaming Connectors
>Reporter: Tzu-Li (Gordon) Tai
>Priority: Critical
>
> Reported in ML: 
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Fink-KafkaProducer-Data-Loss-td11413.html
> The problem:
> The producer holds a {{pendingRecords}} value that is incremented on each 
> invoke() and decremented on each callback, used to check if the producer 
> needs to sync on pending callbacks on checkpoints.
> On each checkpoint, we should only consider the checkpoint succeeded iff 
> after flushing the {{pendingRecords == 0}} and {{asyncException == null}} 
> (currently, we’re only checking {{pendingRecords}}).
> A quick fix for this is to check and rethrow async exceptions in the 
> {{snapshotState}} method both before and after flushing and 
> {{pendingRecords}} becomes 0.



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

[GitHub] flink pull request #3278: [FLINK-5701] [kafka] FlinkKafkaProducer should che...

[tzulitai]

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

https://github.com/apache/flink/pull/3278#discussion_r100036593
  
--- Diff: 
flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBaseTest.java
 ---
@@ -293,6 +293,61 @@ public void run() {
testHarness.close();
}
 
+   /**
+* This test is meant to assure that testAtLeastOnceProducer is valid 
by testing that if flushing is disabled,
+* the snapshot method does indeed finishes without waiting for pending 
records;
+* we set a timeout because the test will not finish if the logic is 
broken
+*/
+   @Test(timeout=5000)
+   public void testDoesNotWaitForPendingRecordsIfFlushingDisabled() throws 
Throwable {
+   final OneShotLatch inputLatch = new OneShotLatch();
+
+   final DummyFlinkKafkaProducer producer = new 
DummyFlinkKafkaProducer<>(
--- End diff --

That makes sense. I was just reusing the {{MockProducer}} implementation in 
this class, which is needed in other tests to be able to provide success / 
failure fake callback completions. In this test we don't need to do that, so a 
simple {{mock(KafkaProducer.class)}} should do.


---
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-5701) FlinkKafkaProducer should check asyncException on checkpoints

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/3278#discussion_r100036011
  
--- Diff: 
flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBaseTest.java
 ---
@@ -293,6 +293,61 @@ public void run() {
testHarness.close();
}
 
+   /**
+* This test is meant to assure that testAtLeastOnceProducer is valid 
by testing that if flushing is disabled,
+* the snapshot method does indeed finishes without waiting for pending 
records;
+* we set a timeout because the test will not finish if the logic is 
broken
+*/
+   @Test(timeout=5000)
+   public void testDoesNotWaitForPendingRecordsIfFlushingDisabled() throws 
Throwable {
+   final OneShotLatch inputLatch = new OneShotLatch();
+
+   final DummyFlinkKafkaProducer producer = new 
DummyFlinkKafkaProducer<>(
+   FakeStandardProducerConfig.get(), null, inputLatch, 
100, new AtomicBoolean(false));
+   producer.setFlushOnCheckpoint(false);
+
+   final OneInputStreamOperatorTestHarness 
testHarness =
+   new OneInputStreamOperatorTestHarness<>(new 
StreamSink(producer));
+
+   testHarness.open();
+
+   List pending = 
producer.getProducerInstance().getPending();
+
+   for (int i = 0; i < 100; i++) {
+   testHarness.processElement(new StreamRecord<>("msg-" + 
i));
+   }
+
+   inputLatch.await();
+
+   // make sure that all callbacks have not been completed
+   Assert.assertEquals(100, pending.size());
+
+   // use a separate thread to continuously monitor whether 
snapshotting has returned
+   final Tuple1 runnableError = new Tuple1<>(null);
+   Thread snapshotThread = new Thread(new Runnable() {
+   @Override
+   public void run() {
+   try {
+   testHarness.snapshot(123L, 123L);
+   } catch (Exception e) {
+   runnableError.f0 = e;
+   }
+   }
+   });
--- End diff --

You're right, not really sure what I was thinking at the time :/ Will 
change this!


> FlinkKafkaProducer should check asyncException on checkpoints
> -
>
> Key: FLINK-5701
> URL: https://issues.apache.org/jira/browse/FLINK-5701
> Project: Flink
>  Issue Type: Bug
>  Components: Kafka Connector, Streaming Connectors
>Reporter: Tzu-Li (Gordon) Tai
>Priority: Critical
>
> Reported in ML: 
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Fink-KafkaProducer-Data-Loss-td11413.html
> The problem:
> The producer holds a {{pendingRecords}} value that is incremented on each 
> invoke() and decremented on each callback, used to check if the producer 
> needs to sync on pending callbacks on checkpoints.
> On each checkpoint, we should only consider the checkpoint succeeded iff 
> after flushing the {{pendingRecords == 0}} and {{asyncException == null}} 
> (currently, we’re only checking {{pendingRecords}}).
> A quick fix for this is to check and rethrow async exceptions in the 
> {{snapshotState}} method both before and after flushing and 
> {{pendingRecords}} becomes 0.



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

[GitHub] flink pull request #3278: [FLINK-5701] [kafka] FlinkKafkaProducer should che...

[tzulitai]

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

https://github.com/apache/flink/pull/3278#discussion_r100036011
  
--- Diff: 
flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBaseTest.java
 ---
@@ -293,6 +293,61 @@ public void run() {
testHarness.close();
}
 
+   /**
+* This test is meant to assure that testAtLeastOnceProducer is valid 
by testing that if flushing is disabled,
+* the snapshot method does indeed finishes without waiting for pending 
records;
+* we set a timeout because the test will not finish if the logic is 
broken
+*/
+   @Test(timeout=5000)
+   public void testDoesNotWaitForPendingRecordsIfFlushingDisabled() throws 
Throwable {
+   final OneShotLatch inputLatch = new OneShotLatch();
+
+   final DummyFlinkKafkaProducer producer = new 
DummyFlinkKafkaProducer<>(
+   FakeStandardProducerConfig.get(), null, inputLatch, 
100, new AtomicBoolean(false));
+   producer.setFlushOnCheckpoint(false);
+
+   final OneInputStreamOperatorTestHarness 
testHarness =
+   new OneInputStreamOperatorTestHarness<>(new 
StreamSink(producer));
+
+   testHarness.open();
+
+   List pending = 
producer.getProducerInstance().getPending();
+
+   for (int i = 0; i < 100; i++) {
+   testHarness.processElement(new StreamRecord<>("msg-" + 
i));
+   }
+
+   inputLatch.await();
+
+   // make sure that all callbacks have not been completed
+   Assert.assertEquals(100, pending.size());
+
+   // use a separate thread to continuously monitor whether 
snapshotting has returned
+   final Tuple1 runnableError = new Tuple1<>(null);
+   Thread snapshotThread = new Thread(new Runnable() {
+   @Override
+   public void run() {
+   try {
+   testHarness.snapshot(123L, 123L);
+   } catch (Exception e) {
+   runnableError.f0 = e;
+   }
+   }
+   });
--- End diff --

You're right, not really sure what I was thinking at the time :/ Will 
change 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-5701) FlinkKafkaProducer should check asyncException on checkpoints

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/3278#discussion_r100035659
  
--- Diff: 
flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBaseTest.java
 ---
@@ -293,6 +293,61 @@ public void run() {
testHarness.close();
}
 
+   /**
+* This test is meant to assure that testAtLeastOnceProducer is valid 
by testing that if flushing is disabled,
+* the snapshot method does indeed finishes without waiting for pending 
records;
+* we set a timeout because the test will not finish if the logic is 
broken
+*/
+   @Test(timeout=5000)
+   public void testDoesNotWaitForPendingRecordsIfFlushingDisabled() throws 
Throwable {
+   final OneShotLatch inputLatch = new OneShotLatch();
+
+   final DummyFlinkKafkaProducer producer = new 
DummyFlinkKafkaProducer<>(
+   FakeStandardProducerConfig.get(), null, inputLatch, 
100, new AtomicBoolean(false));
+   producer.setFlushOnCheckpoint(false);
+
+   final OneInputStreamOperatorTestHarness 
testHarness =
+   new OneInputStreamOperatorTestHarness<>(new 
StreamSink(producer));
+
+   testHarness.open();
+
+   List pending = 
producer.getProducerInstance().getPending();
+
+   for (int i = 0; i < 100; i++) {
+   testHarness.processElement(new StreamRecord<>("msg-" + 
i));
+   }
+
+   inputLatch.await();
--- End diff --

Correct, I'll change this. The other tests might actually not need also, 
I'll check them as well.


> FlinkKafkaProducer should check asyncException on checkpoints
> -
>
> Key: FLINK-5701
> URL: https://issues.apache.org/jira/browse/FLINK-5701
> Project: Flink
>  Issue Type: Bug
>  Components: Kafka Connector, Streaming Connectors
>Reporter: Tzu-Li (Gordon) Tai
>Priority: Critical
>
> Reported in ML: 
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Fink-KafkaProducer-Data-Loss-td11413.html
> The problem:
> The producer holds a {{pendingRecords}} value that is incremented on each 
> invoke() and decremented on each callback, used to check if the producer 
> needs to sync on pending callbacks on checkpoints.
> On each checkpoint, we should only consider the checkpoint succeeded iff 
> after flushing the {{pendingRecords == 0}} and {{asyncException == null}} 
> (currently, we’re only checking {{pendingRecords}}).
> A quick fix for this is to check and rethrow async exceptions in the 
> {{snapshotState}} method both before and after flushing and 
> {{pendingRecords}} becomes 0.



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

[GitHub] flink pull request #3278: [FLINK-5701] [kafka] FlinkKafkaProducer should che...

[tzulitai]

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

https://github.com/apache/flink/pull/3278#discussion_r100035659
  
--- Diff: 
flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBaseTest.java
 ---
@@ -293,6 +293,61 @@ public void run() {
testHarness.close();
}
 
+   /**
+* This test is meant to assure that testAtLeastOnceProducer is valid 
by testing that if flushing is disabled,
+* the snapshot method does indeed finishes without waiting for pending 
records;
+* we set a timeout because the test will not finish if the logic is 
broken
+*/
+   @Test(timeout=5000)
+   public void testDoesNotWaitForPendingRecordsIfFlushingDisabled() throws 
Throwable {
+   final OneShotLatch inputLatch = new OneShotLatch();
+
+   final DummyFlinkKafkaProducer producer = new 
DummyFlinkKafkaProducer<>(
+   FakeStandardProducerConfig.get(), null, inputLatch, 
100, new AtomicBoolean(false));
+   producer.setFlushOnCheckpoint(false);
+
+   final OneInputStreamOperatorTestHarness 
testHarness =
+   new OneInputStreamOperatorTestHarness<>(new 
StreamSink(producer));
+
+   testHarness.open();
+
+   List pending = 
producer.getProducerInstance().getPending();
+
+   for (int i = 0; i < 100; i++) {
+   testHarness.processElement(new StreamRecord<>("msg-" + 
i));
+   }
+
+   inputLatch.await();
--- End diff --

Correct, I'll change this. The other tests might actually not need also, 
I'll check them as well.


---
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 #3278: [FLINK-5701] [kafka] FlinkKafkaProducer should che...

[tzulitai]

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

https://github.com/apache/flink/pull/3278#discussion_r100035474
  
--- Diff: 
flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBaseTest.java
 ---
@@ -293,6 +293,61 @@ public void run() {
testHarness.close();
}
 
+   /**
+* This test is meant to assure that testAtLeastOnceProducer is valid 
by testing that if flushing is disabled,
+* the snapshot method does indeed finishes without waiting for pending 
records;
+* we set a timeout because the test will not finish if the logic is 
broken
+*/
+   @Test(timeout=5000)
+   public void testDoesNotWaitForPendingRecordsIfFlushingDisabled() throws 
Throwable {
+   final OneShotLatch inputLatch = new OneShotLatch();
+
+   final DummyFlinkKafkaProducer producer = new 
DummyFlinkKafkaProducer<>(
+   FakeStandardProducerConfig.get(), null, inputLatch, 
100, new AtomicBoolean(false));
+   producer.setFlushOnCheckpoint(false);
+
+   final OneInputStreamOperatorTestHarness 
testHarness =
+   new OneInputStreamOperatorTestHarness<>(new 
StreamSink(producer));
+
+   testHarness.open();
+
+   List pending = 
producer.getProducerInstance().getPending();
+
+   for (int i = 0; i < 100; i++) {
+   testHarness.processElement(new StreamRecord<>("msg-" + 
i));
+   }
--- End diff --

1 is enough, will change 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-5701) FlinkKafkaProducer should check asyncException on checkpoints

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/3278#discussion_r100035474
  
--- Diff: 
flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBaseTest.java
 ---
@@ -293,6 +293,61 @@ public void run() {
testHarness.close();
}
 
+   /**
+* This test is meant to assure that testAtLeastOnceProducer is valid 
by testing that if flushing is disabled,
+* the snapshot method does indeed finishes without waiting for pending 
records;
+* we set a timeout because the test will not finish if the logic is 
broken
+*/
+   @Test(timeout=5000)
+   public void testDoesNotWaitForPendingRecordsIfFlushingDisabled() throws 
Throwable {
+   final OneShotLatch inputLatch = new OneShotLatch();
+
+   final DummyFlinkKafkaProducer producer = new 
DummyFlinkKafkaProducer<>(
+   FakeStandardProducerConfig.get(), null, inputLatch, 
100, new AtomicBoolean(false));
+   producer.setFlushOnCheckpoint(false);
+
+   final OneInputStreamOperatorTestHarness 
testHarness =
+   new OneInputStreamOperatorTestHarness<>(new 
StreamSink(producer));
+
+   testHarness.open();
+
+   List pending = 
producer.getProducerInstance().getPending();
+
+   for (int i = 0; i < 100; i++) {
+   testHarness.processElement(new StreamRecord<>("msg-" + 
i));
+   }
--- End diff --

1 is enough, will change this.


> FlinkKafkaProducer should check asyncException on checkpoints
> -
>
> Key: FLINK-5701
> URL: https://issues.apache.org/jira/browse/FLINK-5701
> Project: Flink
>  Issue Type: Bug
>  Components: Kafka Connector, Streaming Connectors
>Reporter: Tzu-Li (Gordon) Tai
>Priority: Critical
>
> Reported in ML: 
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Fink-KafkaProducer-Data-Loss-td11413.html
> The problem:
> The producer holds a {{pendingRecords}} value that is incremented on each 
> invoke() and decremented on each callback, used to check if the producer 
> needs to sync on pending callbacks on checkpoints.
> On each checkpoint, we should only consider the checkpoint succeeded iff 
> after flushing the {{pendingRecords == 0}} and {{asyncException == null}} 
> (currently, we’re only checking {{pendingRecords}}).
> A quick fix for this is to check and rethrow async exceptions in the 
> {{snapshotState}} method both before and after flushing and 
> {{pendingRecords}} becomes 0.



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

[jira] [Commented] (FLINK-4364) Implement TaskManager side of heartbeat from JobManager

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/3151#discussion_r100034748
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
 ---
@@ -112,6 +114,13 @@ public TaskManagerRunner(
// Initialize the TM metrics

TaskExecutorMetricsInitializer.instantiateStatusMetrics(taskManagerMetricGroup, 
taskManagerServices.getNetworkEnvironment());
 
+   HeartbeatManagerImpl heartbeatManager = new 
HeartbeatManagerImpl(
+   
taskManagerConfiguration.getTimeout().toMilliseconds(),
+   resourceID,
+   executor,
+   Executors.newSingleThreadScheduledExecutor(),
--- End diff --

Maybe we could add to the `RpcService` that it provides a 
`ScheduledExecutorService`. If we see that the load on this executor is too 
much, then we can introduce a dedicated service. What do you think?


> Implement TaskManager side of heartbeat from JobManager
> ---
>
> Key: FLINK-4364
> URL: https://issues.apache.org/jira/browse/FLINK-4364
> Project: Flink
>  Issue Type: Sub-task
>  Components: Cluster Management
>Reporter: Zhijiang Wang
>Assignee: Zhijiang Wang
>
> The {{JobManager}} initiates heartbeat messages via (JobID, JmLeaderID), and 
> the {{TaskManager}} will report metrics info for each heartbeat.



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

[GitHub] flink pull request #3151: [FLINK-4364][runtime]mplement TaskManager side of ...

[tillrohrmann]

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

https://github.com/apache/flink/pull/3151#discussion_r100034748
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
 ---
@@ -112,6 +114,13 @@ public TaskManagerRunner(
// Initialize the TM metrics

TaskExecutorMetricsInitializer.instantiateStatusMetrics(taskManagerMetricGroup, 
taskManagerServices.getNetworkEnvironment());
 
+   HeartbeatManagerImpl heartbeatManager = new 
HeartbeatManagerImpl(
+   
taskManagerConfiguration.getTimeout().toMilliseconds(),
+   resourceID,
+   executor,
+   Executors.newSingleThreadScheduledExecutor(),
--- End diff --

Maybe we could add to the `RpcService` that it provides a 
`ScheduledExecutorService`. If we see that the load on this executor is too 
much, then we can introduce a dedicated service. What do you think?


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

[jira] [Commented] (FLINK-4364) Implement TaskManager side of heartbeat from JobManager

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/3151#discussion_r100034459
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
 ---
@@ -741,6 +763,18 @@ private void establishJobManagerConnection(JobID 
jobId, JobMasterGateway jobMast
jobManagerTable.put(jobId, 
associateWithJobManager(jobMasterGateway, jobManagerLeaderId, 
registrationSuccess.getBlobPort()));
}
 
+   
heartbeatManager.monitorTarget(registrationSuccess.getResourceID(), new 
HeartbeatTarget() {
+   @Override
+   public void sendHeartbeat(ResourceID resourceID, Object 
payload) {
+   
jobMasterGateway.heartbeatFromTaskManager(resourceID, payload);
+   }
+
+   @Override
+   public void requestHeartbeat(ResourceID resourceID, 
Object payload) {
+   throw new UnsupportedOperationException("Should 
never call requestHeartbeat in task manager.");
--- End diff --

`requestHeartbeat` sends a heartbeat to the target and requests an answer. 
`sendHeartbeat` simply sends a heartbeat to the target. You're right that 
`HeartbeatManagerImpl` won't call `requestHeartbeat`, but the JM should also be 
able to respond to a heartbeat request.


> Implement TaskManager side of heartbeat from JobManager
> ---
>
> Key: FLINK-4364
> URL: https://issues.apache.org/jira/browse/FLINK-4364
> Project: Flink
>  Issue Type: Sub-task
>  Components: Cluster Management
>Reporter: Zhijiang Wang
>Assignee: Zhijiang Wang
>
> The {{JobManager}} initiates heartbeat messages via (JobID, JmLeaderID), and 
> the {{TaskManager}} will report metrics info for each heartbeat.



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

[jira] [Created] (FLINK-5743) Mark WindowedStream.aggregate* methods as PublicEvolving

[Aljoscha Krettek (JIRA)]

Aljoscha Krettek created FLINK-5743:
---

 Summary: Mark WindowedStream.aggregate* methods as PublicEvolving
 Key: FLINK-5743
 URL: https://issues.apache.org/jira/browse/FLINK-5743
 Project: Flink
  Issue Type: Improvement
  Components: DataStream API
Reporter: Aljoscha Krettek
Priority: Blocker


IMHO, they are to new for knowing whether they will persist in their current 
form.  



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

[GitHub] flink pull request #3151: [FLINK-4364][runtime]mplement TaskManager side of ...

[tillrohrmann]

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

https://github.com/apache/flink/pull/3151#discussion_r100034459
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
 ---
@@ -741,6 +763,18 @@ private void establishJobManagerConnection(JobID 
jobId, JobMasterGateway jobMast
jobManagerTable.put(jobId, 
associateWithJobManager(jobMasterGateway, jobManagerLeaderId, 
registrationSuccess.getBlobPort()));
}
 
+   
heartbeatManager.monitorTarget(registrationSuccess.getResourceID(), new 
HeartbeatTarget() {
+   @Override
+   public void sendHeartbeat(ResourceID resourceID, Object 
payload) {
+   
jobMasterGateway.heartbeatFromTaskManager(resourceID, payload);
+   }
+
+   @Override
+   public void requestHeartbeat(ResourceID resourceID, 
Object payload) {
+   throw new UnsupportedOperationException("Should 
never call requestHeartbeat in task manager.");
--- End diff --

`requestHeartbeat` sends a heartbeat to the target and requests an answer. 
`sendHeartbeat` simply sends a heartbeat to the target. You're right that 
`HeartbeatManagerImpl` won't call `requestHeartbeat`, but the JM should also be 
able to respond to a heartbeat request.


---
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 #3285: [FLINK-4997] [streaming] Introduce ProcessWindowFu...

[aljoscha]

GitHub user aljoscha opened a pull request:

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

[FLINK-4997] [streaming] Introduce ProcessWindowFunction

This is an updated/enhanced version of #2756.

I did roughly these changes:
 - Add support for the `AggregatingFunction`/`ProcessWindowFunction` 
combination
 - Mark the new methods/interfaces as `@PublicEvolving`
 - Change how the Scala wrapper functions work because we recently moved 
all of them to use `WrappingFunction`. This is not complete yet and I created 
issues for fixing that: https://issues.apache.org/jira/browse/FLINK-5740 and 
https://issues.apache.org/jira/browse/FLINK-5741
 - Add tests in `WindowTranslationTest.java` and 
`WindowTranslationTest.scala`.

@VenturaDelMonte and @manuzhang, would you like to have a look at this. I 
took quite a while because there was some upheaval in the internals of that 
part of the code and the tests. I'm sorry for that.

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

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

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

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


commit 6f476f92f22b661f26d322615af6a52eca01c69c
Author: Aljoscha Krettek 
Date:   2017-02-07T09:54:54Z

[hotfix] Fix trailing whitespace in WindowedStream.java

commit ab74f4142ddc4aab6e76563b32464d526f719ae7
Author: Ventura Del Monte 
Date:   2016-11-23T17:00:23Z

[FLINK-4997] [streaming] Introduce ProcessWindowFunction

commit b6759f446bad5170fabc803103c764f204412301
Author: Ventura Del Monte 
Date:   2016-11-09T09:49:47Z

[FLINK-4997] [streaming] Add ProcessWindowFunction to Scala API

commit a9791e0e7526bfd419e72ac749ce068f18df857a
Author: Aljoscha Krettek 
Date:   2017-02-07T13:38:25Z

[FLINK-4997] Add ProcessWindowFunction support for .aggregate()

commit 01048feccd5aced42b1979a3bf7b355443c08f5b
Author: Aljoscha Krettek 
Date:   2016-11-24T07:14:48Z

[FLINK-5237] Consolidate and harmonize Window Translation 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-5719) Let LatencyMarkers completely bypass operators / chains

[Robert Metzger (JIRA)]

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

Robert Metzger commented on FLINK-5719:
---

Yes, changing from PublicEvolving to Internal is good!

I don't care so much about changing this, if you think its necessary.
One thing to consider however is the following: If we ever want to add a 
functionality that allows operators to "add" latency, we need to expose the 
latency markers to the operator.
One way to implement this feature would be adding an additional field to the 
marker "addedLatency", where operators can add the milliseconds delay the 
operator is adding.


> Let LatencyMarkers completely bypass operators / chains
> ---
>
> Key: FLINK-5719
> URL: https://issues.apache.org/jira/browse/FLINK-5719
> Project: Flink
>  Issue Type: Improvement
>  Components: DataStream API, Streaming
>Reporter: Tzu-Li (Gordon) Tai
>
> Currently, {{LatencyMarker}} s are forwarded through operators via the 
> operator interfaces and methods, i.e. 
> {{AbstractStreamOperator#processLatencyMarker()}},  
> {{Output#emitLatencyMarker()}}, 
> {{OneInputStreamOperator#processLatencyMarker()}} etc.
> The main issue with this is that {{LatencyMarker}} s are essentially internal 
> elements, and the implementation on how to handle them should be final. 
> Exposing them through operator interfaces will allow the user to override the 
> implementation, and also makes the user interface for operators 
> over-complicated.
> [~aljoscha] suggested to bypass such internal stream elements from the 
> operator to keep the operator interfaces minimal, in FLINK-5017.
> We propose a similar approach here for {{LatencyMarker}} as well. Since the 
> chaining output calls contribute very little to the measured latency and can 
> be ignored, instead of passing it through operator chains, latency markers 
> can simply be passed downstream once tasks receive them.



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

[jira] [Commented] (FLINK-4997) Extending Window Function Metadata

[ASF GitHub Bot (JIRA)]

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

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

GitHub user aljoscha opened a pull request:

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

[FLINK-4997] [streaming] Introduce ProcessWindowFunction

This is an updated/enhanced version of #2756.

I did roughly these changes:
 - Add support for the `AggregatingFunction`/`ProcessWindowFunction` 
combination
 - Mark the new methods/interfaces as `@PublicEvolving`
 - Change how the Scala wrapper functions work because we recently moved 
all of them to use `WrappingFunction`. This is not complete yet and I created 
issues for fixing that: https://issues.apache.org/jira/browse/FLINK-5740 and 
https://issues.apache.org/jira/browse/FLINK-5741
 - Add tests in `WindowTranslationTest.java` and 
`WindowTranslationTest.scala`.

@VenturaDelMonte and @manuzhang, would you like to have a look at this. I 
took quite a while because there was some upheaval in the internals of that 
part of the code and the tests. I'm sorry for that.

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

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

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

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


commit 6f476f92f22b661f26d322615af6a52eca01c69c
Author: Aljoscha Krettek 
Date:   2017-02-07T09:54:54Z

[hotfix] Fix trailing whitespace in WindowedStream.java

commit ab74f4142ddc4aab6e76563b32464d526f719ae7
Author: Ventura Del Monte 
Date:   2016-11-23T17:00:23Z

[FLINK-4997] [streaming] Introduce ProcessWindowFunction

commit b6759f446bad5170fabc803103c764f204412301
Author: Ventura Del Monte 
Date:   2016-11-09T09:49:47Z

[FLINK-4997] [streaming] Add ProcessWindowFunction to Scala API

commit a9791e0e7526bfd419e72ac749ce068f18df857a
Author: Aljoscha Krettek 
Date:   2017-02-07T13:38:25Z

[FLINK-4997] Add ProcessWindowFunction support for .aggregate()

commit 01048feccd5aced42b1979a3bf7b355443c08f5b
Author: Aljoscha Krettek 
Date:   2016-11-24T07:14:48Z

[FLINK-5237] Consolidate and harmonize Window Translation Tests




> Extending Window Function Metadata
> --
>
> Key: FLINK-4997
> URL: https://issues.apache.org/jira/browse/FLINK-4997
> Project: Flink
>  Issue Type: New Feature
>  Components: DataStream API, Streaming
>Reporter: Ventura Del Monte
>Assignee: Ventura Del Monte
>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-2+Extending+Window+Function+Metadata



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

[jira] [Created] (FLINK-5742) Breakpoints on documentation website

[Colin Breame (JIRA)]

Colin Breame created FLINK-5742:
---

 Summary: Breakpoints on documentation website
 Key: FLINK-5742
 URL: https://issues.apache.org/jira/browse/FLINK-5742
 Project: Flink
  Issue Type: Improvement
Affects Versions: 1.2.0
Reporter: Colin Breame
Priority: Trivial


When reading the documentation website, I'm finding that unless I have the 
browser window fully maximised, the content is formatted in the narrow layout.

See https://ci.apache.org/projects/flink/flink-docs-release-1.2/

*Expected behaviour:*
The content should flow on the right-hand-side of the menu.

*Actual behaviour*
The menu is the full width of the window and the content flows below the menu.

*Notes*
Any width below 1200px causes the content to be placed below the full width 
menu.  This 2/3 the width of my laptop screen.

I would suggest making this smaller and consistent with the main project page 
(http://flink.apache.org/) which has the breakpoint set to about 840px.




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

[jira] [Updated] (FLINK-5741) Add tests for window function wrappers with RichFunctions

[Aljoscha Krettek (JIRA)]

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

Aljoscha Krettek updated FLINK-5741:

Priority: Blocker  (was: Major)

> Add tests for window function wrappers with RichFunctions
> -
>
> Key: FLINK-5741
> URL: https://issues.apache.org/jira/browse/FLINK-5741
> Project: Flink
>  Issue Type: Improvement
>  Components: DataStream API
>Reporter: Aljoscha Krettek
>Assignee: Aljoscha Krettek
>Priority: Blocker
>
> There are not tests that {{RichFunctions}} work in all cases and at least 
> {{ScalaProcessWindowFunctionWrapper}} is know to not work correctly before 
> FLINK-5740 is implemented.



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

[jira] [Commented] (FLINK-3885) Revisit parallelism logic

[Till Rohrmann (JIRA)]

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

Till Rohrmann commented on FLINK-3885:
--

I think the problem still exists.

> Revisit parallelism logic
> -
>
> Key: FLINK-3885
> URL: https://issues.apache.org/jira/browse/FLINK-3885
> Project: Flink
>  Issue Type: Bug
>  Components: Streaming
>Affects Versions: 1.1.0
>Reporter: Till Rohrmann
>
> Flink has multiple ways to specify the parallelism of programs. One can set 
> it directly at the operator or job-wide via the {{ExecutionConfig}}. Then it 
> is also possible to define a default parallelism in the {{flink-conf.yaml}}. 
> This default parallelism is only respected on the client side.
> For batch jobs, the default parallelism is consistently set in the 
> {{Optimizer}}. However, for streaming jobs the default parallelism is only 
> set iff the {{StreamExecutionEnvironment}} has been created via 
> {{StreamExecutionEnvironment.getExecutionEnvironment()}}. If one creates a 
> {{RemoteStreamEnvironment}}, then the default parallelism won't be respected. 
> Instead the parallelism is {{-1}}. Also the {{JobManager}} does not respect 
> the default parallelism. But whenever it sees a parallelism of {{-1}} it sets 
> it to {{1}}. This behaviour is not consistent with the batch behaviour and 
> the behaviour described in the documentation of {{parallelism.default}}.
> On top of that we also have the {{PARALLELISM_AUTO_MAX}} value for the 
> parallelism. This value tells the system to use all available slots. However, 
> this feature is nowhere documented in our docs.
> I would propose to get rid of {{PARALLELISM_AUTO_MAX}} and to make the 
> default parallelism behaviour in the batch and streaming API consistent.



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

[jira] [Commented] (FLINK-1743) Add multinomial logistic regression to machine learning library

[Till Rohrmann (JIRA)]

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

Till Rohrmann commented on FLINK-1743:
--

I think [~dedrummond] no longer works on this issue. Thus, I'll assign it to 
you [~mtunqiue].

> Add multinomial logistic regression to machine learning library
> ---
>
> Key: FLINK-1743
> URL: https://issues.apache.org/jira/browse/FLINK-1743
> Project: Flink
>  Issue Type: New Feature
>  Components: Machine Learning Library
>Reporter: Till Rohrmann
>Assignee: David E Drummond
>  Labels: ML
>
> Multinomial logistic regression [1] would be good first classification 
> algorithm which can classify multiple classes. 
> Resources:
> [1] [http://en.wikipedia.org/wiki/Multinomial_logistic_regression]



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

[jira] [Assigned] (FLINK-1743) Add multinomial logistic regression to machine learning library

[Till Rohrmann (JIRA)]

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

Till Rohrmann reassigned FLINK-1743:


Assignee: Tao Meng  (was: David E Drummond)

> Add multinomial logistic regression to machine learning library
> ---
>
> Key: FLINK-1743
> URL: https://issues.apache.org/jira/browse/FLINK-1743
> Project: Flink
>  Issue Type: New Feature
>  Components: Machine Learning Library
>Reporter: Till Rohrmann
>Assignee: Tao Meng
>  Labels: ML
>
> Multinomial logistic regression [1] would be good first classification 
> algorithm which can classify multiple classes. 
> Resources:
> [1] [http://en.wikipedia.org/wiki/Multinomial_logistic_regression]



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

[GitHub] flink issue #3151: [FLINK-4364][runtime]mplement TaskManager side of heartbe...

[wangzhijiang999]

Github user wangzhijiang999 commented on the issue:

https://github.com/apache/flink/pull/3151
  
@tillrohrmann , I am preparing the testing code and can submit the updates 
this week. Thank you for continuous help!


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