[jira] [Commented] (FLINK-6738) HBaseConnectorITCase is flaky

[Ted Yu (JIRA)]

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

Ted Yu commented on FLINK-6738:
---

Calcite 1.14 has been released.

> HBaseConnectorITCase is flaky
> -
>
> Key: FLINK-6738
> URL: https://issues.apache.org/jira/browse/FLINK-6738
> Project: Flink
>  Issue Type: Bug
>  Components: Streaming Connectors
>Reporter: Ted Yu
>Priority: Critical
>  Labels: hbase, test-stability
>
> I ran integration tests for flink 1.3 RC2 and got the following failure:
> {code}
> Failed tests:
>   
> HBaseConnectorITCase>HBaseTestingClusterAutostarter.tearDown:240->HBaseTestingClusterAutostarter.deleteTables:127
>  Exception found deleting the table expected null, but 
> was: java.util.concurrent.TimeoutException: The procedure 5 is still running>
> {code}



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (FLINK-7432) Unclosed HighAvailabilityServices instance in QueryableStateClient

[Ted Yu (JIRA)]

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

Ted Yu updated FLINK-7432:
--
Description: 
{code}
  public QueryableStateClient(Configuration config) throws Exception {
this(config, HighAvailabilityServicesUtils.createHighAvailabilityServices(
config, Executors.directExecutor(), 
HighAvailabilityServicesUtils.AddressResolution.TRY_ADDRESS_RESOLUTION));
{code}
The HighAvailabilityServices instance is only used for calling 
getJobManagerLeaderRetriever().

The instance should be closed upon leaving QueryableStateClient ctor.

  was:
{code}
  public QueryableStateClient(Configuration config) throws Exception {
this(config, HighAvailabilityServicesUtils.createHighAvailabilityServices(
config, Executors.directExecutor(), 
HighAvailabilityServicesUtils.AddressResolution.TRY_ADDRESS_RESOLUTION));
{code}

The HighAvailabilityServices instance is only used for calling 
getJobManagerLeaderRetriever().

The instance should be closed upon leaving QueryableStateClient ctor.


> Unclosed HighAvailabilityServices instance in QueryableStateClient
> --
>
> Key: FLINK-7432
> URL: https://issues.apache.org/jira/browse/FLINK-7432
> Project: Flink
>  Issue Type: Bug
>  Components: Queryable State
>Reporter: Ted Yu
>Priority: Minor
>
> {code}
>   public QueryableStateClient(Configuration config) throws Exception {
> this(config, HighAvailabilityServicesUtils.createHighAvailabilityServices(
> config, Executors.directExecutor(), 
> HighAvailabilityServicesUtils.AddressResolution.TRY_ADDRESS_RESOLUTION));
> {code}
> The HighAvailabilityServices instance is only used for calling 
> getJobManagerLeaderRetriever().
> The instance should be closed upon leaving QueryableStateClient ctor.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-6094) Implement stream-stream proctime non-window inner join

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/4471#discussion_r143320024
  
--- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamJoinRule.scala
 ---
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.plan.rules.datastream
+
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.table.api.TableConfig
+import org.apache.flink.table.calcite.FlinkTypeFactory
+import org.apache.flink.table.plan.nodes.FlinkConventions
+import org.apache.flink.table.plan.nodes.datastream.DataStreamJoin
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalJoin
+import org.apache.flink.table.plan.schema.RowSchema
+import org.apache.flink.table.runtime.join.WindowJoinUtil
+import scala.collection.JavaConverters._
+
+class DataStreamJoinRule
+  extends ConverterRule(
+classOf[FlinkLogicalJoin],
+FlinkConventions.LOGICAL,
+FlinkConventions.DATASTREAM,
+"DataStreamJoinRule") {
+
+  override def matches(call: RelOptRuleCall): Boolean = {
+val join: FlinkLogicalJoin = call.rel(0).asInstanceOf[FlinkLogicalJoin]
+val joinInfo = join.analyzeCondition
+
+val (windowBounds, remainingPreds) = 
WindowJoinUtil.extractWindowBoundsFromPredicate(
+  joinInfo.getRemaining(join.getCluster.getRexBuilder),
+  join.getLeft.getRowType.getFieldCount,
+  join.getRowType,
+  join.getCluster.getRexBuilder,
+  TableConfig.DEFAULT)
+
+// remaining predicate must not access time attributes
+val remainingPredsAccessTime = remainingPreds.isDefined &&
+  WindowJoinUtil.accessesTimeAttribute(remainingPreds.get, 
join.getRowType)
+
+// Check that no event-time attributes are in the input.
+val rowTimeAttrInOutput = join.getRowType.getFieldList.asScala
+  .exists(f => FlinkTypeFactory.isRowtimeIndicatorType(f.getType))
+
+if (!windowBounds.isDefined && !remainingPredsAccessTime && 
!rowTimeAttrInOutput) {
--- End diff --

Hi @fhueske , thanks a lot for this review. Maybe we need several 
iterations before everything is ok. I will update the PR and address your 
comments(except for this one). We can have more discussions then. I will update 
the PR in one or two days. 
Thanks, Hequn


> Implement stream-stream proctime non-window  inner join
> ---
>
> Key: FLINK-6094
> URL: https://issues.apache.org/jira/browse/FLINK-6094
> Project: Flink
>  Issue Type: New Feature
>  Components: Table API & SQL
>Reporter: Shaoxuan Wang
>Assignee: Hequn Cheng
>
> This includes:
> 1.Implement stream-stream proctime non-window  inner join
> 2.Implement the retract process logic for join



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink pull request #4471: [FLINK-6094] [table] Implement stream-stream proct...

[hequn8128]

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

https://github.com/apache/flink/pull/4471#discussion_r143320024
  
--- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamJoinRule.scala
 ---
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.plan.rules.datastream
+
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.table.api.TableConfig
+import org.apache.flink.table.calcite.FlinkTypeFactory
+import org.apache.flink.table.plan.nodes.FlinkConventions
+import org.apache.flink.table.plan.nodes.datastream.DataStreamJoin
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalJoin
+import org.apache.flink.table.plan.schema.RowSchema
+import org.apache.flink.table.runtime.join.WindowJoinUtil
+import scala.collection.JavaConverters._
+
+class DataStreamJoinRule
+  extends ConverterRule(
+classOf[FlinkLogicalJoin],
+FlinkConventions.LOGICAL,
+FlinkConventions.DATASTREAM,
+"DataStreamJoinRule") {
+
+  override def matches(call: RelOptRuleCall): Boolean = {
+val join: FlinkLogicalJoin = call.rel(0).asInstanceOf[FlinkLogicalJoin]
+val joinInfo = join.analyzeCondition
+
+val (windowBounds, remainingPreds) = 
WindowJoinUtil.extractWindowBoundsFromPredicate(
+  joinInfo.getRemaining(join.getCluster.getRexBuilder),
+  join.getLeft.getRowType.getFieldCount,
+  join.getRowType,
+  join.getCluster.getRexBuilder,
+  TableConfig.DEFAULT)
+
+// remaining predicate must not access time attributes
+val remainingPredsAccessTime = remainingPreds.isDefined &&
+  WindowJoinUtil.accessesTimeAttribute(remainingPreds.get, 
join.getRowType)
+
+// Check that no event-time attributes are in the input.
+val rowTimeAttrInOutput = join.getRowType.getFieldList.asScala
+  .exists(f => FlinkTypeFactory.isRowtimeIndicatorType(f.getType))
+
+if (!windowBounds.isDefined && !remainingPredsAccessTime && 
!rowTimeAttrInOutput) {
--- End diff --

Hi @fhueske , thanks a lot for this review. Maybe we need several 
iterations before everything is ok. I will update the PR and address your 
comments(except for this one). We can have more discussions then. I will update 
the PR in one or two days. 
Thanks, Hequn


---

[jira] [Commented] (FLINK-3655) Allow comma-separated or multiple directories to be specified for FileInputFormat

[Vishnu Viswanath (JIRA)]

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

Vishnu Viswanath commented on FLINK-3655:
-

was looking for this feature. why wasn't this ever merged?

> Allow comma-separated or multiple directories to be specified for 
> FileInputFormat
> -
>
> Key: FLINK-3655
> URL: https://issues.apache.org/jira/browse/FLINK-3655
> Project: Flink
>  Issue Type: Improvement
>  Components: Core
>Affects Versions: 1.0.0
>Reporter: Gna Phetsarath
>Priority: Minor
>  Labels: starter
>
> Allow comma-separated or multiple directories to be specified for 
> FileInputFormat so that a DataSource will process the directories 
> sequentially.
>
> env.readFile("/data/2016/01/01/*/*,/data/2016/01/02/*/*,/data/2016/01/03/*/*")
> in Scala
>env.readFile(paths: Seq[String])
> or 
>   env.readFile(path: String, otherPaths: String*)
> Wildcard support would be a bonus.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-7312) activate checkstyle for flink/core/memory/*

[ASF GitHub Bot (JIRA)]

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

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

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4447
  
#4445 contains this PR, yes.


> activate checkstyle for flink/core/memory/*
> ---
>
> Key: FLINK-7312
> URL: https://issues.apache.org/jira/browse/FLINK-7312
> Project: Flink
>  Issue Type: Sub-task
>  Components: Checkstyle, Core
>Affects Versions: 1.4.0
>Reporter: Nico Kruber
>Assignee: Nico Kruber
>Priority: Minor
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink issue #4447: [FLINK-7312][checkstyle] activate checkstyle for flink/co...

[zentol]

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4447
  
#4445 contains this PR, yes.


---

[jira] [Resolved] (FLINK-7768) Load File Systems via Java Service abstraction

[Stephan Ewen (JIRA)]

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

Stephan Ewen resolved FLINK-7768.
-
  Resolution: Fixed
Assignee: Stephan Ewen
Release Note: Implemented in 77e3701ca1f8bfab33a07f11992955eb131126c3

> Load File Systems via Java Service abstraction
> --
>
> Key: FLINK-7768
> URL: https://issues.apache.org/jira/browse/FLINK-7768
> Project: Flink
>  Issue Type: Improvement
>  Components: Core
>Reporter: Stephan Ewen
>Assignee: Stephan Ewen
> Fix For: 1.4.0
>
>
> We should change the discovery mechanism of file from static class name 
> configurations to a service mechanism (META-INF/services). 
> With this change, users can add new filesystem implementations and make them 
> available by simply adding them to the class path. 
> As part of that, factoring HDFS and MapR FS implementations into separate 
> modules helps with a better and more fine grained dependency management, 
> needing less explicit reflection logic.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (FLINK-7768) Load File Systems via Java Service abstraction

[Stephan Ewen (JIRA)]

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

Stephan Ewen updated FLINK-7768:

Description: 
We should change the discovery mechanism of file from static class name 
configurations to a service mechanism (META-INF/services). 

With this change, users can add new filesystem implementations and make them 
available by simply adding them to the class path. 

As part of that, factoring HDFS and MapR FS implementations into separate 
modules helps with a better and more fine grained dependency management, 
needing less explicit reflection logic.

> Load File Systems via Java Service abstraction
> --
>
> Key: FLINK-7768
> URL: https://issues.apache.org/jira/browse/FLINK-7768
> Project: Flink
>  Issue Type: Improvement
>  Components: Core
>Reporter: Stephan Ewen
> Fix For: 1.4.0
>
>
> We should change the discovery mechanism of file from static class name 
> configurations to a service mechanism (META-INF/services). 
> With this change, users can add new filesystem implementations and make them 
> available by simply adding them to the class path. 
> As part of that, factoring HDFS and MapR FS implementations into separate 
> modules helps with a better and more fine grained dependency management, 
> needing less explicit reflection logic.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (FLINK-7768) Load File Systems via Java Service abstraction

[Stephan Ewen (JIRA)]

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

Stephan Ewen updated FLINK-7768:

Release Note:   (was: We should change the discovery mechanism of file from 
static class name configurations to a service mechanism (META-INF/services).

With this change, users can add new filesystem implementations and make them 
available by simply adding them to the class path.

As part of that, factoring HDFS and MapR FS implementations into separate 
modules helps with a better and more fine grained dependency management, 
needing less explicit reflection logic.)

> Load File Systems via Java Service abstraction
> --
>
> Key: FLINK-7768
> URL: https://issues.apache.org/jira/browse/FLINK-7768
> Project: Flink
>  Issue Type: Improvement
>  Components: Core
>Reporter: Stephan Ewen
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Closed] (FLINK-7768) Load File Systems via Java Service abstraction

[Stephan Ewen (JIRA)]

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

Stephan Ewen closed FLINK-7768.
---

> Load File Systems via Java Service abstraction
> --
>
> Key: FLINK-7768
> URL: https://issues.apache.org/jira/browse/FLINK-7768
> Project: Flink
>  Issue Type: Improvement
>  Components: Core
>Reporter: Stephan Ewen
>Assignee: Stephan Ewen
> Fix For: 1.4.0
>
>
> We should change the discovery mechanism of file from static class name 
> configurations to a service mechanism (META-INF/services). 
> With this change, users can add new filesystem implementations and make them 
> available by simply adding them to the class path. 
> As part of that, factoring HDFS and MapR FS implementations into separate 
> modules helps with a better and more fine grained dependency management, 
> needing less explicit reflection logic.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-7312) activate checkstyle for flink/core/memory/*

[ASF GitHub Bot (JIRA)]

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

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

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/4447
  
This seems to be subsumed by #4445 - is that correct?


> activate checkstyle for flink/core/memory/*
> ---
>
> Key: FLINK-7312
> URL: https://issues.apache.org/jira/browse/FLINK-7312
> Project: Flink
>  Issue Type: Sub-task
>  Components: Checkstyle, Core
>Affects Versions: 1.4.0
>Reporter: Nico Kruber
>Assignee: Nico Kruber
>Priority: Minor
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink issue #4447: [FLINK-7312][checkstyle] activate checkstyle for flink/co...

[StephanEwen]

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/4447
  
This seems to be subsumed by #4445 - is that correct?


---

[GitHub] flink issue #4445: [FLINK-7310][core] always use the HybridMemorySegment

[StephanEwen]

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/4445
  
Agree with @KurtYoung.
Merging this...


---

[jira] [Commented] (FLINK-7310) always use HybridMemorySegment

[ASF GitHub Bot (JIRA)]

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

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

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/4445
  
Agree with @KurtYoung.
Merging this...


> always use HybridMemorySegment
> --
>
> Key: FLINK-7310
> URL: https://issues.apache.org/jira/browse/FLINK-7310
> Project: Flink
>  Issue Type: Sub-task
>  Components: Core
>Affects Versions: 1.4.0
>Reporter: Nico Kruber
>Assignee: Nico Kruber
>
> For future changes to the network buffers (sending our own off-heap buffers 
> through to netty), we cannot use {{HeapMemorySegment}} anymore and need to 
> rely on {{HybridMemorySegment}} instead.
> We should thus drop any code that loads the {{HeapMemorySegment}} (it is 
> still available if needed) in favour of the {{HybridMemorySegment}} which is 
> able to work on both heap and off-heap memory.
> FYI: For the performance penalty of this change compared to using 
> {{HeapMemorySegment}} alone, see this interesting blob article (from 2015):
> https://flink.apache.org/news/2015/09/16/off-heap-memory.html



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (FLINK-7495) AbstractUdfStreamOperator#initializeState() should be called in AsyncWaitOperator#initializeState()

[Robert Metzger (JIRA)]

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

Robert Metzger updated FLINK-7495:
--
Component/s: State Backends, Checkpointing

> AbstractUdfStreamOperator#initializeState() should be called in 
> AsyncWaitOperator#initializeState()
> ---
>
> Key: FLINK-7495
> URL: https://issues.apache.org/jira/browse/FLINK-7495
> Project: Flink
>  Issue Type: Bug
>  Components: State Backends, Checkpointing
>Reporter: Ted Yu
>Assignee: Fang Yong
>Priority: Minor
>
> {code}
> recoveredStreamElements = context
>   .getOperatorStateStore()
>   .getListState(new ListStateDescriptor<>(STATE_NAME, 
> inStreamElementSerializer));
> {code}
> Call to AbstractUdfStreamOperator#initializeState() should be added in the 
> beginning



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (FLINK-7488) TaskManagerHeapSizeCalculationJavaBashTest sometimes fails

[Robert Metzger (JIRA)]

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

Robert Metzger updated FLINK-7488:
--
Component/s: Tests

> TaskManagerHeapSizeCalculationJavaBashTest sometimes fails
> --
>
> Key: FLINK-7488
> URL: https://issues.apache.org/jira/browse/FLINK-7488
> Project: Flink
>  Issue Type: Test
>  Components: Tests
>Reporter: Ted Yu
>Priority: Minor
>
> {code}
> compareNetworkBufShellScriptWithJava(org.apache.flink.dist.TaskManagerHeapSizeCalculationJavaBashTest)
>   Time elapsed: 0.239 sec  <<< FAILURE!
> org.junit.ComparisonFailure: Different network buffer memory sizes with 
> configuration: {taskmanager.network.memory.fraction=0.1, 
> taskmanager.memory.off-heap=false, taskmanager.memory.fraction=0.7, 
> taskmanager.memory.size=-1, taskmanager.network.memory.max=1073741824, 
> taskmanager.heap.mb=1000, taskmanager.network.memory.min=67108864} 
> expected:<[]104857600> but was:<[Setting HADOOP_CONF_DIR=/etc/hadoop/conf 
> because no HADOOP_CONF_DIR was set.Using the result of 'hadoop classpath' to 
> augment the Hadoop classpath: 
> /usr/hdp/2.5.0.0-1245/hadoop/conf:/usr/hdp/2.5.0.0-1245/hadoop/lib/*:/usr/hdp/2.5.0.0-1245/hadoop/.//*:/usr/hdp/2.5.0.0-1245/hadoop-hdfs/./:/usr/hdp/2.5.0.0-1245/hadoop-hdfs/lib/*:/usr/hdp/2.5.0.0-1245/hadoop-hdfs/.//*:/usr/hdp/2.5.0.0-1245/hadoop-yarn/lib/*:/usr/hdp/2.5.0.0-1245/hadoop-yarn/.//*:/usr/hdp/2.5.0.0-1245/hadoop-mapreduce/lib/*:/usr/hdp/2.5.0.0-1245/hadoop-mapreduce/.//*:/usr/hdp/2.5.0.0-1245/tez/*:/usr/hdp/2.5.0.0-1245/tez/lib/*:/usr/hdp/2.5.0.0-1245/tez/conf]104857600>
>   at org.junit.Assert.assertEquals(Assert.java:115)
>   at 
> org.apache.flink.dist.TaskManagerHeapSizeCalculationJavaBashTest.compareNetworkBufJavaVsScript(TaskManagerHeapSizeCalculationJavaBashTest.java:235)
>   at 
> org.apache.flink.dist.TaskManagerHeapSizeCalculationJavaBashTest.compareNetworkBufShellScriptWithJava(TaskManagerHeapSizeCalculationJavaBashTest.java:81)
> compareHeapSizeShellScriptWithJava(org.apache.flink.dist.TaskManagerHeapSizeCalculationJavaBashTest)
>   Time elapsed: 0.16 sec  <<< FAILURE!
> org.junit.ComparisonFailure: Different heap sizes with configuration: 
> {taskmanager.network.memory.fraction=0.1, taskmanager.memory.off-heap=false, 
> taskmanager.memory.fraction=0.7, taskmanager.memory.size=-1, 
> taskmanager.network.memory.max=1073741824, taskmanager.heap.mb=1000, 
> taskmanager.network.memory.min=67108864} expected:<[]1000> but was:<[Setting 
> HADOOP_CONF_DIR=/etc/hadoop/conf because no HADOOP_CONF_DIR was set.Using the 
> result of 'hadoop classpath' to augment the Hadoop classpath: 
> /usr/hdp/2.5.0.0-1245/hadoop/conf:/usr/hdp/2.5.0.0-1245/hadoop/lib/*:/usr/hdp/2.5.0.0-1245/hadoop/.//*:/usr/hdp/2.5.0.0-1245/hadoop-hdfs/./:/usr/hdp/2.5.0.0-1245/hadoop-hdfs/lib/*:/usr/hdp/2.5.0.0-1245/hadoop-hdfs/.//*:/usr/hdp/2.5.0.0-1245/hadoop-yarn/lib/*:/usr/hdp/2.5.0.0-1245/hadoop-yarn/.//*:/usr/hdp/2.5.0.0-1245/hadoop-mapreduce/lib/*:/usr/hdp/2.5.0.0-1245/hadoop-mapreduce/.//*:/usr/hdp/2.5.0.0-1245/tez/*:/usr/hdp/2.5.0.0-1245/tez/lib/*:/usr/hdp/2.5.0.0-1245/tez/conf]1000>
>   at org.junit.Assert.assertEquals(Assert.java:115)
>   at 
> org.apache.flink.dist.TaskManagerHeapSizeCalculationJavaBashTest.compareHeapSizeJavaVsScript(TaskManagerHeapSizeCalculationJavaBashTest.java:275)
>   at 
> org.apache.flink.dist.TaskManagerHeapSizeCalculationJavaBashTest.compareHeapSizeShellScriptWithJava(TaskManagerHeapSizeCalculationJavaBashTest.java:110)
> {code}
> $HADOOP_CONF_DIR was not set prior to running the test.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (FLINK-7775) Remove unreferenced method PermanentBlobCache#getNumberOfCachedJobs

[Robert Metzger (JIRA)]

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

Robert Metzger updated FLINK-7775:
--
Component/s: Local Runtime

> Remove unreferenced method PermanentBlobCache#getNumberOfCachedJobs
> ---
>
> Key: FLINK-7775
> URL: https://issues.apache.org/jira/browse/FLINK-7775
> Project: Flink
>  Issue Type: Task
>  Components: Local Runtime
>Reporter: Ted Yu
>Priority: Minor
>
> {code}
>   public int getNumberOfCachedJobs() {
> return jobRefCounters.size();
>   }
> {code}
> The method is not used.
> We should remove it.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (FLINK-7740) Add parameter support in CassandraInputFormat

[Robert Metzger (JIRA)]

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

Robert Metzger updated FLINK-7740:
--
Component/s: Cassandra Connector

> Add parameter support in CassandraInputFormat
> -
>
> Key: FLINK-7740
> URL: https://issues.apache.org/jira/browse/FLINK-7740
> Project: Flink
>  Issue Type: Improvement
>  Components: Cassandra Connector
>Reporter: Bin Wang
>Priority: Minor
>
> I suggest to add a small improvement of CassandraInputFormat. It support CQL 
> string as input only. I think adding parameter support is good for both 
> security and make the CQL string simpler when there is a IN clause or the 
> parameter value is very long.
> e.g. "SELECT col0, col1, col2 from keyspace.table0 where col0 in ? and col1>?"
> vs "SELECT col0, col1, col2 from keyspace.table0 where col0 in ('v0', 
> 'v1','v2',...,'vn') and col > 12345678901234"



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (FLINK-7682) Invalid state size formatting

[Robert Metzger (JIRA)]

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

Robert Metzger updated FLINK-7682:
--
Component/s: Webfrontend

> Invalid state size formatting
> -
>
> Key: FLINK-7682
> URL: https://issues.apache.org/jira/browse/FLINK-7682
> Project: Flink
>  Issue Type: Bug
>  Components: Webfrontend
>Affects Versions: 1.3.0
>Reporter: Andrey
>Priority: Minor
>
> Steps to reproduce:
> * open running job
> * click "checkpoints" tab
> * Last completed checkpoint has "State Size: 1.00e+3 MB"
> Expected:
> "State Size: 1 GB" or "State Size: 1000 MB"



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (FLINK-7642) Upgrade maven surefire plugin to 2.19.1

[Robert Metzger (JIRA)]

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

Robert Metzger updated FLINK-7642:
--
Component/s: Build System

> Upgrade maven surefire plugin to 2.19.1
> ---
>
> Key: FLINK-7642
> URL: https://issues.apache.org/jira/browse/FLINK-7642
> Project: Flink
>  Issue Type: Improvement
>  Components: Build System
>Reporter: Ted Yu
>
> Surefire 2.19 release introduced more useful test filters which would let us 
> run a subset of the test.
> This issue is for upgrading maven surefire plugin to 2.19.1



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-7493) Create Pinot Connector

[Robert Metzger (JIRA)]

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

Robert Metzger commented on FLINK-7493:
---

I would suggest to contribute the connector to the Apache Bahir project.

> Create Pinot Connector
> --
>
> Key: FLINK-7493
> URL: https://issues.apache.org/jira/browse/FLINK-7493
> Project: Flink
>  Issue Type: New Feature
>  Components: Streaming Connectors
>Reporter: Zhenqiu Huang
>Assignee: Zhenqiu Huang
>
> Add pinot connector for streaming ingestion and batch segment file push.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-7492) Memsql Connector

[Robert Metzger (JIRA)]

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

Robert Metzger commented on FLINK-7492:
---

I would suggest to contribute the connector to the Apache Bahir project.

> Memsql Connector
> 
>
> Key: FLINK-7492
> URL: https://issues.apache.org/jira/browse/FLINK-7492
> Project: Flink
>  Issue Type: New Feature
>  Components: Streaming Connectors
>Reporter: Zhenqiu Huang
>Assignee: Zhenqiu Huang
>
> Add an output connector for both streaming and batch ingestion for Memsql. 



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (FLINK-7679) Upgrade maven enforcer plugin to 3.0.0-M1

[Robert Metzger (JIRA)]

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

Robert Metzger updated FLINK-7679:
--
Component/s: Build System

> Upgrade maven enforcer plugin to 3.0.0-M1
> -
>
> Key: FLINK-7679
> URL: https://issues.apache.org/jira/browse/FLINK-7679
> Project: Flink
>  Issue Type: Bug
>  Components: Build System
>Reporter: Ted Yu
>
> I got the following build error against Java 9:
> {code}
> [ERROR] Failed to execute goal 
> org.apache.maven.plugins:maven-enforcer-plugin:1.4.1:enforce (enforce-maven) 
> on project flink-parent: Execution enforce-maven of goal 
> org.apache.maven.plugins:maven-enforcer-plugin:1.4.1:enforce failed: An API 
> incompatibility was encountered while executing 
> org.apache.maven.plugins:maven-enforcer-plugin:1.4.1:enforce: 
> java.lang.ExceptionInInitializerError: null
> [ERROR] -
> [ERROR] realm =plugin>org.apache.maven.plugins:maven-enforcer-plugin:1.4.1
> [ERROR] strategy = org.codehaus.plexus.classworlds.strategy.SelfFirstStrategy
> [ERROR] urls[0] = 
> file:/home/hbase/.m2/repository/org/apache/maven/plugins/maven-enforcer-plugin/1.4.1/maven-enforcer-plugin-1.4.1.jar
> {code}
> Upgrading maven enforcer plugin to 3.0.0-M1 would get over the above error.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (FLINK-7493) Create Pinot Connector

[Robert Metzger (JIRA)]

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

Robert Metzger updated FLINK-7493:
--
Component/s: Streaming Connectors

> Create Pinot Connector
> --
>
> Key: FLINK-7493
> URL: https://issues.apache.org/jira/browse/FLINK-7493
> Project: Flink
>  Issue Type: New Feature
>  Components: Streaming Connectors
>Reporter: Zhenqiu Huang
>Assignee: Zhenqiu Huang
>
> Add pinot connector for streaming ingestion and batch segment file push.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (FLINK-7743) Remove the restriction of minimum memory of JM

[Robert Metzger (JIRA)]

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

Robert Metzger updated FLINK-7743:
--
Component/s: YARN

> Remove the restriction of minimum memory of JM
> --
>
> Key: FLINK-7743
> URL: https://issues.apache.org/jira/browse/FLINK-7743
> Project: Flink
>  Issue Type: Bug
>  Components: YARN
>Reporter: Haohui Mai
>Assignee: Haohui Mai
>
> Per discussion on 
> http://mail-archives.apache.org/mod_mbox/flink-user/201709.mbox/%3c4f77255e-1ddb-4e99-a667-73941b110...@apache.org%3E
> It might be great to remove the restriction of the minimum heap size of the 
> JM.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (FLINK-7492) Memsql Connector

[Robert Metzger (JIRA)]

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

Robert Metzger updated FLINK-7492:
--
Component/s: Streaming Connectors

> Memsql Connector
> 
>
> Key: FLINK-7492
> URL: https://issues.apache.org/jira/browse/FLINK-7492
> Project: Flink
>  Issue Type: New Feature
>  Components: Streaming Connectors
>Reporter: Zhenqiu Huang
>Assignee: Zhenqiu Huang
>
> Add an output connector for both streaming and batch ingestion for Memsql. 



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Created] (FLINK-7775) Remove unreferenced method PermanentBlobCache#getNumberOfCachedJobs

[Ted Yu (JIRA)]

Ted Yu created FLINK-7775:
-

 Summary: Remove unreferenced method 
PermanentBlobCache#getNumberOfCachedJobs
 Key: FLINK-7775
 URL: https://issues.apache.org/jira/browse/FLINK-7775
 Project: Flink
  Issue Type: Task
Reporter: Ted Yu
Priority: Minor


{code}
  public int getNumberOfCachedJobs() {
return jobRefCounters.size();
  }
{code}
The method is not used.

We should remove it.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-7744) Add missing top links to documentation

[ASF GitHub Bot (JIRA)]

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

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

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/4756
  
Good from my side.
Since I am not very opinionated about docs/navigation, would be good to 
have a second opinion.

@alpinegizmo what do you think?


> Add missing top links to documentation
> --
>
> Key: FLINK-7744
> URL: https://issues.apache.org/jira/browse/FLINK-7744
> Project: Flink
>  Issue Type: Improvement
>  Components: Documentation
>Reporter: Sirko B
>Priority: Minor
>
> The links to the top of the page are missing on many pages. Those are very 
> useful for reading the documentation on mobile.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink issue #4755: [hotfix] [docs] Fix broken links

[StephanEwen]

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/4755
  
+1, merging this...


---

[GitHub] flink issue #4756: [FLINK-7744][docs] Add missing top links to documentation

[StephanEwen]

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/4756
  
Good from my side.
Since I am not very opinionated about docs/navigation, would be good to 
have a second opinion.

@alpinegizmo what do you think?


---

[jira] [Commented] (FLINK-7768) Load File Systems via Java Service abstraction

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> Load File Systems via Java Service abstraction
> --
>
> Key: FLINK-7768
> URL: https://issues.apache.org/jira/browse/FLINK-7768
> Project: Flink
>  Issue Type: Improvement
>  Components: Core
>Reporter: Stephan Ewen
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink pull request #4781: [FLINK-7768] [core] Load File Systems via Java Ser...

[asfgit]

Github user asfgit closed the pull request at:

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


---

[GitHub] flink issue #4754: [FLINK-7742][hotfix] Fix array access might be out of bou...

[StephanEwen]

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/4754
  
Good catches, thanks.

Merging this...


---

[jira] [Commented] (FLINK-7742) Fix array access might be out of bounds

[ASF GitHub Bot (JIRA)]

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

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

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/4754
  
Good catches, thanks.

Merging this...


> Fix array access might be out of bounds
> ---
>
> Key: FLINK-7742
> URL: https://issues.apache.org/jira/browse/FLINK-7742
> Project: Flink
>  Issue Type: Bug
>  Components: Build System
>Affects Versions: 1.3.2
>Reporter: Hai Zhou UTC+8
>Assignee: Hai Zhou UTC+8
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-7739) Improve Kafka*ITCase tests stability

[ASF GitHub Bot (JIRA)]

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

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

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/4775
  
Thanks, will merge this without the added restart delay. If it is still 
unstable, we can add that back.


> Improve Kafka*ITCase tests stability
> 
>
> Key: FLINK-7739
> URL: https://issues.apache.org/jira/browse/FLINK-7739
> Project: Flink
>  Issue Type: Improvement
>  Components: Kafka Connector
>Affects Versions: 1.3.2
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink issue #4775: [FLINK-7739] Fix KafkaXXITCase tests stability

[StephanEwen]

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/4775
  
Thanks, will merge this without the added restart delay. If it is still 
unstable, we can add that back.


---

[jira] [Commented] (FLINK-7739) Improve Kafka*ITCase tests stability

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/4775#discussion_r143254179
  
--- Diff: 
flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java
 ---
@@ -121,10 +122,12 @@ public static void shutDownServices() throws 
Exception {
 
protected static Configuration getFlinkConfiguration() {
Configuration flinkConfig = new Configuration();
+   flinkConfig.setString(AkkaOptions.WATCH_HEARTBEAT_PAUSE, "5 s");
+   flinkConfig.setString(AkkaOptions.WATCH_HEARTBEAT_INTERVAL, "1 
s");

flinkConfig.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUM_TMS);

flinkConfig.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, TM_SLOTS);
flinkConfig.setLong(TaskManagerOptions.MANAGED_MEMORY_SIZE, 
16L);
-   
flinkConfig.setString(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY, "0 
s");
+   
flinkConfig.setString(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY, "5 
s");
--- End diff --

If we can avoid this, we will save time during testing


> Improve Kafka*ITCase tests stability
> 
>
> Key: FLINK-7739
> URL: https://issues.apache.org/jira/browse/FLINK-7739
> Project: Flink
>  Issue Type: Improvement
>  Components: Kafka Connector
>Affects Versions: 1.3.2
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink pull request #4775: [FLINK-7739] Fix KafkaXXITCase tests stability

[StephanEwen]

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

https://github.com/apache/flink/pull/4775#discussion_r143254179
  
--- Diff: 
flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java
 ---
@@ -121,10 +122,12 @@ public static void shutDownServices() throws 
Exception {
 
protected static Configuration getFlinkConfiguration() {
Configuration flinkConfig = new Configuration();
+   flinkConfig.setString(AkkaOptions.WATCH_HEARTBEAT_PAUSE, "5 s");
+   flinkConfig.setString(AkkaOptions.WATCH_HEARTBEAT_INTERVAL, "1 
s");

flinkConfig.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUM_TMS);

flinkConfig.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, TM_SLOTS);
flinkConfig.setLong(TaskManagerOptions.MANAGED_MEMORY_SIZE, 
16L);
-   
flinkConfig.setString(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY, "0 
s");
+   
flinkConfig.setString(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY, "5 
s");
--- End diff --

If we can avoid this, we will save time during testing


---

[GitHub] flink pull request #4777: [FLINK-7739] Enable dependency convergence

[StephanEwen]

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

https://github.com/apache/flink/pull/4777#discussion_r143253601
  
--- Diff: flink-core/pom.xml ---
@@ -77,6 +88,12 @@ under the License.

org.apache.commons
commons-compress
+   
+   
--- End diff --

This seems to be an exclusion without replacement. By accident?


---

[jira] [Commented] (FLINK-7739) Improve Kafka*ITCase tests stability

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/4777#discussion_r143253601
  
--- Diff: flink-core/pom.xml ---
@@ -77,6 +88,12 @@ under the License.

org.apache.commons
commons-compress
+   
+   
--- End diff --

This seems to be an exclusion without replacement. By accident?


> Improve Kafka*ITCase tests stability
> 
>
> Key: FLINK-7739
> URL: https://issues.apache.org/jira/browse/FLINK-7739
> Project: Flink
>  Issue Type: Improvement
>  Components: Kafka Connector
>Affects Versions: 1.3.2
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-7739) Improve Kafka*ITCase tests stability

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/4777#discussion_r143252870
  
--- Diff: flink-core/pom.xml ---
@@ -63,9 +63,20 @@ under the License.

com.esotericsoftware.kryo
kryo
+   
--- End diff --

I think if you add a dependency in the same pom (as below) you don't need 
the exclusion any more.


> Improve Kafka*ITCase tests stability
> 
>
> Key: FLINK-7739
> URL: https://issues.apache.org/jira/browse/FLINK-7739
> Project: Flink
>  Issue Type: Improvement
>  Components: Kafka Connector
>Affects Versions: 1.3.2
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-7739) Improve Kafka*ITCase tests stability

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/4777#discussion_r143253445
  
--- Diff: pom.xml ---
@@ -1181,7 +1204,7 @@ under the License.

org.apache.maven.plugins
maven-enforcer-plugin
-   1.4.1
+   
${enforcer.version}
--- End diff --

How about defining the plugin in the plugin management section, rather than 
using a variable? I think that is the more maintainable solution in the log run.


> Improve Kafka*ITCase tests stability
> 
>
> Key: FLINK-7739
> URL: https://issues.apache.org/jira/browse/FLINK-7739
> Project: Flink
>  Issue Type: Improvement
>  Components: Kafka Connector
>Affects Versions: 1.3.2
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink pull request #4777: [FLINK-7739] Enable dependency convergence

[StephanEwen]

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

https://github.com/apache/flink/pull/4777#discussion_r143252870
  
--- Diff: flink-core/pom.xml ---
@@ -63,9 +63,20 @@ under the License.

com.esotericsoftware.kryo
kryo
+   
--- End diff --

I think if you add a dependency in the same pom (as below) you don't need 
the exclusion any more.


---

[GitHub] flink pull request #4777: [FLINK-7739] Enable dependency convergence

[StephanEwen]

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

https://github.com/apache/flink/pull/4777#discussion_r143253445
  
--- Diff: pom.xml ---
@@ -1181,7 +1204,7 @@ under the License.

org.apache.maven.plugins
maven-enforcer-plugin
-   1.4.1
+   
${enforcer.version}
--- End diff --

How about defining the plugin in the plugin management section, rather than 
using a variable? I think that is the more maintainable solution in the log run.


---

[jira] [Commented] (FLINK-7266) Don't attempt to delete parent directory on S3

[Stephan Ewen (JIRA)]

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

Stephan Ewen commented on FLINK-7266:
-

True, this is a problem in 1.3.2 - the tradeoff was to either have a very large 
amount of redundant requests for directory emptiness check (which cause the 
checkpointing to stall or be throttled) or to leave the "directories".

In Flink 1.4 we want to fix this by letting the checkpoints understand the file 
structure and make it a single call to drop the directory, as Steve suggested.
The current abstraction is overly generic (just things in arbitrary byte 
chunks) and does not understand that checkpoint files cluster together in 
directories.

> Don't attempt to delete parent directory on S3
> --
>
> Key: FLINK-7266
> URL: https://issues.apache.org/jira/browse/FLINK-7266
> Project: Flink
>  Issue Type: Bug
>  Components: Core
>Affects Versions: 1.3.1
>Reporter: Stephan Ewen
>Assignee: Stephan Ewen
>Priority: Critical
> Fix For: 1.4.0, 1.3.2
>
>
> Currently, every attempted release of an S3 state object also checks if the 
> "parent directory" is empty and then tries to delete it.
> Not only is that unnecessary on S3, but it is prohibitively expensive and for 
> example causes S3 to throttle calls by the JobManager on checkpoint cleanup.
> The {{FileState}} must only attempt parent directory cleanup when operating 
> against real file systems, not when operating against object stores.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink issue #4783: [FLINK-7774][network] fix not clearing deserializers on c...

[StephanEwen]

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/4783
  
Looks good, +1 to merge this


---

[GitHub] flink issue #4782: [FLINK-7772][blob] fix test instability in BlobCacheDelet...

[StephanEwen]

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/4782
  
Thanks for fixing this, merging...


---

[jira] [Commented] (FLINK-7772) Test instability in BlobCacheDeleteTest

[ASF GitHub Bot (JIRA)]

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

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

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/4782
  
Thanks for fixing this, merging...


> Test instability in BlobCacheDeleteTest
> ---
>
> Key: FLINK-7772
> URL: https://issues.apache.org/jira/browse/FLINK-7772
> Project: Flink
>  Issue Type: Bug
>  Components: Network
>Affects Versions: 1.4.0
>Reporter: Nico Kruber
>Assignee: Nico Kruber
>Priority: Critical
>  Labels: test-stability
>
> {{BlobCacheDeleteTest}} did not account for the server executing the delete 
> call of a transient BLOB after acknowledging the request. This resulted in 
> the {{testDeleteTransientLocalFails*}} tests failing.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-7774) Deserializers are not cleaned up when closing input streams

[ASF GitHub Bot (JIRA)]

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

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

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/4783
  
Looks good, +1 to merge this


> Deserializers are not cleaned up when closing input streams
> ---
>
> Key: FLINK-7774
> URL: https://issues.apache.org/jira/browse/FLINK-7774
> Project: Flink
>  Issue Type: Bug
>  Components: Network
>Affects Versions: 1.4.0, 1.3.2
>Reporter: Nico Kruber
>Assignee: Nico Kruber
>
> On cleanup of the {{AbstractRecordReader}}, {{StreamInputProcessor}}, and 
> {{StreamTwoInputProcessor}}, the deserializers' current buffers are cleaned 
> up but not their internal {{spanningWrapper}} and {{nonSpanningWrapper}} via 
> {{RecordDeserializer#clear}}. This call should be added.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-7761) Twitter example is not self-contained

[ASF GitHub Bot (JIRA)]

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

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

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/4773
  
Looks good to me!


> Twitter example is not self-contained
> -
>
> Key: FLINK-7761
> URL: https://issues.apache.org/jira/browse/FLINK-7761
> Project: Flink
>  Issue Type: Bug
>  Components: Examples
>Affects Versions: 1.4.0
>Reporter: Chesnay Schepler
>Assignee: Chesnay Schepler
> Fix For: 1.4.0
>
>
> The Twitter example jar is not self-contained as it excludes the shaded guava 
> dependency from the twitter connector.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-6495) Migrate Akka configuration options

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/4774#discussion_r143246933
  
--- Diff: 
flink-core/src/main/java/org/apache/flink/configuration/IllegalConfigurationException.java
 ---
@@ -42,6 +42,17 @@ public IllegalConfigurationException(String message) {
 
/**
 * Constructs an new IllegalConfigurationException with the given error 
message
+* format and arguments.
+*
+* @param format The error message format for the exception.
+* @param arguments The arguments for the format.
+*/
+   public IllegalConfigurationException(String format, Object... 
arguments) {
--- End diff --

Curious: Why introduce the extra constructor and not call 
`String.format(...)` where the exception is created?


> Migrate Akka configuration options
> --
>
> Key: FLINK-6495
> URL: https://issues.apache.org/jira/browse/FLINK-6495
> Project: Flink
>  Issue Type: Sub-task
>  Components: Distributed Coordination, Network
>Reporter: Chesnay Schepler
>Assignee: Fang Yong
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink issue #4773: [FLINK-7761] [examples] Include shaded guava dependency i...

[StephanEwen]

Github user StephanEwen commented on the issue:

https://github.com/apache/flink/pull/4773
  
Looks good to me!


---

[GitHub] flink pull request #4774: [FLINK-6495] Fix Akka's default value for heartbea...

[StephanEwen]

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

https://github.com/apache/flink/pull/4774#discussion_r143248194
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/restart/FixedDelayRestartStrategy.java
 ---
@@ -82,7 +82,7 @@ public static FixedDelayRestartStrategyFactory 
createFactory(Configuration confi
int maxAttempts = 
configuration.getInteger(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, 
1);
 
String timeoutString = configuration.getString(
-   AkkaOptions.WATCH_HEARTBEAT_INTERVAL);
+   AkkaOptions.WATCH_HEARTBEAT_PAUSE);
--- End diff --

We cannot make this change, this introduces crazy delay on each recovery.


---

[jira] [Commented] (FLINK-6495) Migrate Akka configuration options

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/4774#discussion_r143248194
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/restart/FixedDelayRestartStrategy.java
 ---
@@ -82,7 +82,7 @@ public static FixedDelayRestartStrategyFactory 
createFactory(Configuration confi
int maxAttempts = 
configuration.getInteger(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, 
1);
 
String timeoutString = configuration.getString(
-   AkkaOptions.WATCH_HEARTBEAT_INTERVAL);
+   AkkaOptions.WATCH_HEARTBEAT_PAUSE);
--- End diff --

We cannot make this change, this introduces crazy delay on each recovery.


> Migrate Akka configuration options
> --
>
> Key: FLINK-6495
> URL: https://issues.apache.org/jira/browse/FLINK-6495
> Project: Flink
>  Issue Type: Sub-task
>  Components: Distributed Coordination, Network
>Reporter: Chesnay Schepler
>Assignee: Fang Yong
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink pull request #4774: [FLINK-6495] Fix Akka's default value for heartbea...

[StephanEwen]

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

https://github.com/apache/flink/pull/4774#discussion_r143246933
  
--- Diff: 
flink-core/src/main/java/org/apache/flink/configuration/IllegalConfigurationException.java
 ---
@@ -42,6 +42,17 @@ public IllegalConfigurationException(String message) {
 
/**
 * Constructs an new IllegalConfigurationException with the given error 
message
+* format and arguments.
+*
+* @param format The error message format for the exception.
+* @param arguments The arguments for the format.
+*/
+   public IllegalConfigurationException(String format, Object... 
arguments) {
--- End diff --

Curious: Why introduce the extra constructor and not call 
`String.format(...)` where the exception is created?


---

[jira] [Closed] (FLINK-4660) HadoopFileSystem (with S3A) may leak connections, which cause job to stuck in a restarting loop

[Stephan Ewen (JIRA)]

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

Stephan Ewen closed FLINK-4660.
---
  Resolution: Fixed
Release Note: Closing the reopened issue to update versions in which this 
is fixed

> HadoopFileSystem (with S3A) may leak connections, which cause job to stuck in 
> a restarting loop
> ---
>
> Key: FLINK-4660
> URL: https://issues.apache.org/jira/browse/FLINK-4660
> Project: Flink
>  Issue Type: Bug
>  Components: State Backends, Checkpointing
>Reporter: Zhenzhong Xu
>Priority: Critical
> Fix For: 1.4.0, 1.3.2
>
> Attachments: Screen Shot 2016-09-20 at 2.49.14 PM.png, Screen Shot 
> 2016-09-20 at 2.49.32 PM.png
>
>
> Flink job with checkpoints enabled and configured to use S3A file system 
> backend, sometimes experiences checkpointing failure due to S3 consistency 
> issue. This behavior is also reported by other people and documented in 
> https://issues.apache.org/jira/browse/FLINK-4218.
> This problem gets magnified by current HadoopFileSystem implementation, which 
> can potentially leak S3 client connections, and eventually get into a 
> restarting loop with “Timeout waiting for a connection from pool” exception 
> thrown from aws client.
> I looked at the code, seems HadoopFileSystem.java never invoke close() method 
> on fs object upon failure, but the FileSystem may be re-initialized every 
> time the job gets restarted.
> A few evidence I observed:
> 1. When I set the connection pool limit to 128, and below commands shows 128 
> connections are stuck in CLOSE_WAIT state.
> !Screen Shot 2016-09-20 at 2.49.14 PM.png|align=left, vspace=5! 
> 2. task manager logs indicates that state backend file system consistently 
> getting initialized upon job restarting.
> !Screen Shot 2016-09-20 at 2.49.32 PM.png!
> 3. Log indicates there is NPE during cleanning up of stream task which was 
> caused by “Timeout waiting for connection from pool” exception when trying to 
> create a directory in S3 bucket.
> 2016-09-02 08:17:50,886 ERROR 
> org.apache.flink.streaming.runtime.tasks.StreamTask - Error during cleanup of 
> stream task
> java.lang.NullPointerException
> at 
> org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.cleanup(OneInputStreamTask.java:73)
> at 
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:323)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:589)
> at java.lang.Thread.run(Thread.java:745)
> 4.It appears StreamTask from invoking checkpointing operation, to handling 
> failure, there is no logic associated with closing Hadoop File System object 
> (which internally includes S3 aws client object), which resides in 
> HadoopFileSystem.java.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (FLINK-4660) HadoopFileSystem (with S3A) may leak connections, which cause job to stuck in a restarting loop

[Stephan Ewen (JIRA)]

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

Stephan Ewen updated FLINK-4660:

Fix Version/s: 1.3.2
   1.4.0

> HadoopFileSystem (with S3A) may leak connections, which cause job to stuck in 
> a restarting loop
> ---
>
> Key: FLINK-4660
> URL: https://issues.apache.org/jira/browse/FLINK-4660
> Project: Flink
>  Issue Type: Bug
>  Components: State Backends, Checkpointing
>Reporter: Zhenzhong Xu
>Priority: Critical
> Fix For: 1.4.0, 1.3.2
>
> Attachments: Screen Shot 2016-09-20 at 2.49.14 PM.png, Screen Shot 
> 2016-09-20 at 2.49.32 PM.png
>
>
> Flink job with checkpoints enabled and configured to use S3A file system 
> backend, sometimes experiences checkpointing failure due to S3 consistency 
> issue. This behavior is also reported by other people and documented in 
> https://issues.apache.org/jira/browse/FLINK-4218.
> This problem gets magnified by current HadoopFileSystem implementation, which 
> can potentially leak S3 client connections, and eventually get into a 
> restarting loop with “Timeout waiting for a connection from pool” exception 
> thrown from aws client.
> I looked at the code, seems HadoopFileSystem.java never invoke close() method 
> on fs object upon failure, but the FileSystem may be re-initialized every 
> time the job gets restarted.
> A few evidence I observed:
> 1. When I set the connection pool limit to 128, and below commands shows 128 
> connections are stuck in CLOSE_WAIT state.
> !Screen Shot 2016-09-20 at 2.49.14 PM.png|align=left, vspace=5! 
> 2. task manager logs indicates that state backend file system consistently 
> getting initialized upon job restarting.
> !Screen Shot 2016-09-20 at 2.49.32 PM.png!
> 3. Log indicates there is NPE during cleanning up of stream task which was 
> caused by “Timeout waiting for connection from pool” exception when trying to 
> create a directory in S3 bucket.
> 2016-09-02 08:17:50,886 ERROR 
> org.apache.flink.streaming.runtime.tasks.StreamTask - Error during cleanup of 
> stream task
> java.lang.NullPointerException
> at 
> org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.cleanup(OneInputStreamTask.java:73)
> at 
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:323)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:589)
> at java.lang.Thread.run(Thread.java:745)
> 4.It appears StreamTask from invoking checkpointing operation, to handling 
> failure, there is no logic associated with closing Hadoop File System object 
> (which internally includes S3 aws client object), which resides in 
> HadoopFileSystem.java.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Reopened] (FLINK-4660) HadoopFileSystem (with S3A) may leak connections, which cause job to stuck in a restarting loop

[Stephan Ewen (JIRA)]

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

Stephan Ewen reopened FLINK-4660:
-

Reopening to assign proper fix version

> HadoopFileSystem (with S3A) may leak connections, which cause job to stuck in 
> a restarting loop
> ---
>
> Key: FLINK-4660
> URL: https://issues.apache.org/jira/browse/FLINK-4660
> Project: Flink
>  Issue Type: Bug
>  Components: State Backends, Checkpointing
>Reporter: Zhenzhong Xu
>Priority: Critical
> Attachments: Screen Shot 2016-09-20 at 2.49.14 PM.png, Screen Shot 
> 2016-09-20 at 2.49.32 PM.png
>
>
> Flink job with checkpoints enabled and configured to use S3A file system 
> backend, sometimes experiences checkpointing failure due to S3 consistency 
> issue. This behavior is also reported by other people and documented in 
> https://issues.apache.org/jira/browse/FLINK-4218.
> This problem gets magnified by current HadoopFileSystem implementation, which 
> can potentially leak S3 client connections, and eventually get into a 
> restarting loop with “Timeout waiting for a connection from pool” exception 
> thrown from aws client.
> I looked at the code, seems HadoopFileSystem.java never invoke close() method 
> on fs object upon failure, but the FileSystem may be re-initialized every 
> time the job gets restarted.
> A few evidence I observed:
> 1. When I set the connection pool limit to 128, and below commands shows 128 
> connections are stuck in CLOSE_WAIT state.
> !Screen Shot 2016-09-20 at 2.49.14 PM.png|align=left, vspace=5! 
> 2. task manager logs indicates that state backend file system consistently 
> getting initialized upon job restarting.
> !Screen Shot 2016-09-20 at 2.49.32 PM.png!
> 3. Log indicates there is NPE during cleanning up of stream task which was 
> caused by “Timeout waiting for connection from pool” exception when trying to 
> create a directory in S3 bucket.
> 2016-09-02 08:17:50,886 ERROR 
> org.apache.flink.streaming.runtime.tasks.StreamTask - Error during cleanup of 
> stream task
> java.lang.NullPointerException
> at 
> org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.cleanup(OneInputStreamTask.java:73)
> at 
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:323)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:589)
> at java.lang.Thread.run(Thread.java:745)
> 4.It appears StreamTask from invoking checkpointing operation, to handling 
> failure, there is no logic associated with closing Hadoop File System object 
> (which internally includes S3 aws client object), which resides in 
> HadoopFileSystem.java.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-4660) HadoopFileSystem (with S3A) may leak connections, which cause job to stuck in a restarting loop

[Stephan Ewen (JIRA)]

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

Stephan Ewen commented on FLINK-4660:
-

This was part of the 1.3.2 release and will be part of the 1.4.0 release.

> HadoopFileSystem (with S3A) may leak connections, which cause job to stuck in 
> a restarting loop
> ---
>
> Key: FLINK-4660
> URL: https://issues.apache.org/jira/browse/FLINK-4660
> Project: Flink
>  Issue Type: Bug
>  Components: State Backends, Checkpointing
>Reporter: Zhenzhong Xu
>Priority: Critical
> Attachments: Screen Shot 2016-09-20 at 2.49.14 PM.png, Screen Shot 
> 2016-09-20 at 2.49.32 PM.png
>
>
> Flink job with checkpoints enabled and configured to use S3A file system 
> backend, sometimes experiences checkpointing failure due to S3 consistency 
> issue. This behavior is also reported by other people and documented in 
> https://issues.apache.org/jira/browse/FLINK-4218.
> This problem gets magnified by current HadoopFileSystem implementation, which 
> can potentially leak S3 client connections, and eventually get into a 
> restarting loop with “Timeout waiting for a connection from pool” exception 
> thrown from aws client.
> I looked at the code, seems HadoopFileSystem.java never invoke close() method 
> on fs object upon failure, but the FileSystem may be re-initialized every 
> time the job gets restarted.
> A few evidence I observed:
> 1. When I set the connection pool limit to 128, and below commands shows 128 
> connections are stuck in CLOSE_WAIT state.
> !Screen Shot 2016-09-20 at 2.49.14 PM.png|align=left, vspace=5! 
> 2. task manager logs indicates that state backend file system consistently 
> getting initialized upon job restarting.
> !Screen Shot 2016-09-20 at 2.49.32 PM.png!
> 3. Log indicates there is NPE during cleanning up of stream task which was 
> caused by “Timeout waiting for connection from pool” exception when trying to 
> create a directory in S3 bucket.
> 2016-09-02 08:17:50,886 ERROR 
> org.apache.flink.streaming.runtime.tasks.StreamTask - Error during cleanup of 
> stream task
> java.lang.NullPointerException
> at 
> org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.cleanup(OneInputStreamTask.java:73)
> at 
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:323)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:589)
> at java.lang.Thread.run(Thread.java:745)
> 4.It appears StreamTask from invoking checkpointing operation, to handling 
> failure, there is no logic associated with closing Hadoop File System object 
> (which internally includes S3 aws client object), which resides in 
> HadoopFileSystem.java.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Closed] (FLINK-7767) Avoid loading Hadoop conf dynamically at runtime

[Stephan Ewen (JIRA)]

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

Stephan Ewen closed FLINK-7767.
---

> Avoid loading Hadoop conf dynamically at runtime
> 
>
> Key: FLINK-7767
> URL: https://issues.apache.org/jira/browse/FLINK-7767
> Project: Flink
>  Issue Type: Improvement
>  Components: Streaming Connectors
>Reporter: Stephan Ewen
>Assignee: Stephan Ewen
> Fix For: 1.4.0
>
>
> The bucketing sink dynamically loads the Hadoop configuration in various 
> places.
> The result of that configuration is not always predictable, as it tries to 
> automagically discover the Hadoop config files.
> A better approach is to rely on the Flink configuration to find the Hadoop 
> configuration, or to directly use the Hadoop configuration used by the Hadoop 
> file systems.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Resolved] (FLINK-7767) Avoid loading Hadoop conf dynamically at runtime

[Stephan Ewen (JIRA)]

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

Stephan Ewen resolved FLINK-7767.
-
  Resolution: Fixed
Release Note: Fixed via bad3df54d20677157f48c3ee1f3251d2c4bce8ba

> Avoid loading Hadoop conf dynamically at runtime
> 
>
> Key: FLINK-7767
> URL: https://issues.apache.org/jira/browse/FLINK-7767
> Project: Flink
>  Issue Type: Improvement
>  Components: Streaming Connectors
>Reporter: Stephan Ewen
>Assignee: Stephan Ewen
> Fix For: 1.4.0
>
>
> The bucketing sink dynamically loads the Hadoop configuration in various 
> places.
> The result of that configuration is not always predictable, as it tries to 
> automagically discover the Hadoop config files.
> A better approach is to rely on the Flink configuration to find the Hadoop 
> configuration, or to directly use the Hadoop configuration used by the Hadoop 
> file systems.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Resolved] (FLINK-7766) Remove obsolete reflection for hflush on HDFS

[Stephan Ewen (JIRA)]

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

Stephan Ewen resolved FLINK-7766.
-
  Resolution: Fixed
Release Note: Fixed via 7843c2ffb44f99967dc71746ac1c79b04a74fe80

> Remove obsolete reflection for hflush on HDFS
> -
>
> Key: FLINK-7766
> URL: https://issues.apache.org/jira/browse/FLINK-7766
> Project: Flink
>  Issue Type: Improvement
>  Components: Streaming Connectors
>Reporter: Stephan Ewen
>Assignee: Stephan Ewen
> Fix For: 1.4.0
>
>
> This code originally existed for compatibility with Hadoop 1.
> Since Hadoop 1 support is dropped, this is no longer necessary.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Closed] (FLINK-7766) Remove obsolete reflection for hflush on HDFS

[Stephan Ewen (JIRA)]

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

Stephan Ewen closed FLINK-7766.
---

> Remove obsolete reflection for hflush on HDFS
> -
>
> Key: FLINK-7766
> URL: https://issues.apache.org/jira/browse/FLINK-7766
> Project: Flink
>  Issue Type: Improvement
>  Components: Streaming Connectors
>Reporter: Stephan Ewen
>Assignee: Stephan Ewen
> Fix For: 1.4.0
>
>
> This code originally existed for compatibility with Hadoop 1.
> Since Hadoop 1 support is dropped, this is no longer necessary.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-7766) Remove obsolete reflection for hflush on HDFS

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> Remove obsolete reflection for hflush on HDFS
> -
>
> Key: FLINK-7766
> URL: https://issues.apache.org/jira/browse/FLINK-7766
> Project: Flink
>  Issue Type: Improvement
>  Components: Streaming Connectors
>Reporter: Stephan Ewen
>Assignee: Stephan Ewen
> Fix For: 1.4.0
>
>
> This code originally existed for compatibility with Hadoop 1.
> Since Hadoop 1 support is dropped, this is no longer necessary.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink pull request #4780: [FLINK-7766] [FLINK-7767] [file system sink] Clean...

[asfgit]

Github user asfgit closed the pull request at:

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


---

[jira] [Commented] (FLINK-2973) Add flink-benchmark with compliant licenses again

[Fabian Hueske (JIRA)]

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

Fabian Hueske commented on FLINK-2973:
--

Not sure. 
[This page|http://www.apache.org/licenses/GPL-compatibility.html] suggests that 
even linking to GPL licensed software is not allowed because "it is considered 
by the GPLv3 authors to create a derivative work". 

> Add flink-benchmark with compliant licenses again
> -
>
> Key: FLINK-2973
> URL: https://issues.apache.org/jira/browse/FLINK-2973
> Project: Flink
>  Issue Type: Task
>  Components: Build System
>Affects Versions: 1.0.0
>Reporter: Fabian Hueske
>Assignee: Suneel Marthi
>Priority: Minor
> Fix For: 1.0.0
>
>
> We recently created the Maven module {{flink-benchmark}} for micro-benchmarks 
> and ported most of the existing micro-benchmarks to the Java benchmarking 
> framework JMH. However, JMH is part of OpenJDK and under GPL license which is 
> not compatible with the AL2.
> Consequently, we need to remove this dependency and either revert the porting 
> commits or port the benchmarks to another benchmarking framework. An 
> alternative could be [Google's Caliper|https://github.com/google/caliper] 
> library which is under AL2.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink pull request #4625: [FLINK-6233] [table] Support time-bounded stream i...

[xccui]

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

https://github.com/apache/flink/pull/4625#discussion_r143231150
  
--- Diff: 
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/sql/JoinITCase.scala
 ---
@@ -102,5 +117,154 @@ class JoinITCase extends StreamingWithStateTestBase {
 env.execute()
   }
 
+  /** test rowtime inner join **/
+  @Test
+  def testRowTimeInnerJoin(): Unit = {
+val env = StreamExecutionEnvironment.getExecutionEnvironment
+val tEnv = TableEnvironment.getTableEnvironment(env)
+env.setStateBackend(getStateBackend)
+env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+StreamITCase.clear
+env.setParallelism(1)
+
+val sqlQuery =
+  """
+|SELECT t2.a, t2.c, t1.c
+|FROM T1 as t1 join T2 as t2 ON
+|  t1.a = t2.a AND
+|  t1.rt BETWEEN t2.rt - INTERVAL '5' SECOND AND
+|t2.rt + INTERVAL '6' SECOND
+|""".stripMargin
+
+val data1 = new mutable.MutableList[(Int, Long, String, Long)]
--- End diff --

If I understand correctly, this problem was caused by the different 
semantics of `Null` in SQL and other common languages (i.e., `Null =? Null`). 
We transform the equi-conditions to a `keyBy` operation and maybe rows with 
`Null` keys are mapped to the same group, thus being taken as identical?


---

[jira] [Commented] (FLINK-6233) Support rowtime inner equi-join between two streams in the SQL API

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/4625#discussion_r143231150
  
--- Diff: 
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/sql/JoinITCase.scala
 ---
@@ -102,5 +117,154 @@ class JoinITCase extends StreamingWithStateTestBase {
 env.execute()
   }
 
+  /** test rowtime inner join **/
+  @Test
+  def testRowTimeInnerJoin(): Unit = {
+val env = StreamExecutionEnvironment.getExecutionEnvironment
+val tEnv = TableEnvironment.getTableEnvironment(env)
+env.setStateBackend(getStateBackend)
+env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+StreamITCase.clear
+env.setParallelism(1)
+
+val sqlQuery =
+  """
+|SELECT t2.a, t2.c, t1.c
+|FROM T1 as t1 join T2 as t2 ON
+|  t1.a = t2.a AND
+|  t1.rt BETWEEN t2.rt - INTERVAL '5' SECOND AND
+|t2.rt + INTERVAL '6' SECOND
+|""".stripMargin
+
+val data1 = new mutable.MutableList[(Int, Long, String, Long)]
--- End diff --

If I understand correctly, this problem was caused by the different 
semantics of `Null` in SQL and other common languages (i.e., `Null =? Null`). 
We transform the equi-conditions to a `keyBy` operation and maybe rows with 
`Null` keys are mapped to the same group, thus being taken as identical?


> Support rowtime inner equi-join between two streams in the SQL API
> --
>
> Key: FLINK-6233
> URL: https://issues.apache.org/jira/browse/FLINK-6233
> Project: Flink
>  Issue Type: Sub-task
>  Components: Table API & SQL
>Reporter: hongyuhong
>Assignee: Xingcan Cui
>
> The goal of this issue is to add support for inner equi-join on proc time 
> streams to the SQL interface.
> Queries similar to the following should be supported:
> {code}
> SELECT o.rowtime , o.productId, o.orderId, s.rowtime AS shipTime 
> FROM Orders AS o 
> JOIN Shipments AS s 
> ON o.orderId = s.orderId 
> AND o.rowtime BETWEEN s.rowtime AND s.rowtime + INTERVAL '1' HOUR;
> {code}
> The following restrictions should initially apply:
> * The join hint only support inner join
> * The ON clause should include equi-join condition
> * The time-condition {{o.rowtime BETWEEN s.rowtime AND s.rowtime + INTERVAL 
> '1' HOUR}} only can use rowtime that is a system attribute, the time 
> condition only support bounded time range like {{o.rowtime BETWEEN s.rowtime 
> - INTERVAL '1' HOUR AND s.rowtime + INTERVAL '1' HOUR}}, not support 
> unbounded like {{o.rowtime  s.rowtime}} ,  and  should include both two 
> stream's rowtime attribute, {{o.rowtime between rowtime () and rowtime () + 
> 1}} should also not be supported.
> An row-time streams join will not be able to handle late data, because this 
> would mean in insert a row into a sorted order shift all other computations. 
> This would be too expensive to maintain. Therefore, we will throw an error if 
> a user tries to use an row-time stream join with late data handling.
> This issue includes:
> * Design of the DataStream operator to deal with stream join
> * Translation from Calcite's RelNode representation (LogicalJoin). 



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Assigned] (FLINK-7548) Support watermark generation for TableSource

[Fabian Hueske (JIRA)]

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

Fabian Hueske reassigned FLINK-7548:


Assignee: Fabian Hueske

> Support watermark generation for TableSource
> 
>
> Key: FLINK-7548
> URL: https://issues.apache.org/jira/browse/FLINK-7548
> Project: Flink
>  Issue Type: Bug
>  Components: Table API & SQL
>Reporter: Jark Wu
>Assignee: Fabian Hueske
>
> As discussed in FLINK-7446, currently the TableSource only support to define 
> rowtime field, but not support to extract watermarks from the rowtime field. 
> We can provide a new interface called {{DefinedWatermark}}, which has two 
> methods {{getRowtimeAttribute}} (can only be an existing field) and 
> {{getWatermarkGenerator}}. The {{DefinedRowtimeAttribute}} will be marked 
> deprecated.
> How to support periodic and punctuated watermarks and support some built-in 
> strategies needs further discussion.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-6233) Support rowtime inner equi-join between two streams in the SQL API

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/4625#discussion_r143222605
  
--- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamWindowJoin.scala
 ---
@@ -184,4 +229,50 @@ class DataStreamWindowJoin(
 .returns(returnTypeInfo)
 }
   }
+
+  def createRowTimeInnerJoin(
+  leftDataStream: DataStream[CRow],
+  rightDataStream: DataStream[CRow],
+  returnTypeInfo: TypeInformation[CRow],
+  joinFunctionName: String,
+  joinFunctionCode: String,
+  leftKeys: Array[Int],
+  rightKeys: Array[Int]): DataStream[CRow] = {
+
+val rowTimeInnerJoinFunc = new RowTimeBoundedStreamInnerJoin(
+  leftLowerBound,
+  leftUpperBound,
+  allowedLateness = 0L,
+  leftSchema.typeInfo,
+  rightSchema.typeInfo,
+  joinFunctionName,
+  joinFunctionCode,
+  leftTimeIdx,
+  rightTimeIdx)
+
+if (!leftKeys.isEmpty) {
+  leftDataStream
+.connect(rightDataStream)
+.keyBy(leftKeys, rightKeys)
--- End diff --

Actually, I was quite confused about this condition `!leftKeys.isEmpty`. 
Since in `FlinkLogicalJoin.scala`, queries without equi-conditions are blocked, 
when will this condition be hold?


> Support rowtime inner equi-join between two streams in the SQL API
> --
>
> Key: FLINK-6233
> URL: https://issues.apache.org/jira/browse/FLINK-6233
> Project: Flink
>  Issue Type: Sub-task
>  Components: Table API & SQL
>Reporter: hongyuhong
>Assignee: Xingcan Cui
>
> The goal of this issue is to add support for inner equi-join on proc time 
> streams to the SQL interface.
> Queries similar to the following should be supported:
> {code}
> SELECT o.rowtime , o.productId, o.orderId, s.rowtime AS shipTime 
> FROM Orders AS o 
> JOIN Shipments AS s 
> ON o.orderId = s.orderId 
> AND o.rowtime BETWEEN s.rowtime AND s.rowtime + INTERVAL '1' HOUR;
> {code}
> The following restrictions should initially apply:
> * The join hint only support inner join
> * The ON clause should include equi-join condition
> * The time-condition {{o.rowtime BETWEEN s.rowtime AND s.rowtime + INTERVAL 
> '1' HOUR}} only can use rowtime that is a system attribute, the time 
> condition only support bounded time range like {{o.rowtime BETWEEN s.rowtime 
> - INTERVAL '1' HOUR AND s.rowtime + INTERVAL '1' HOUR}}, not support 
> unbounded like {{o.rowtime  s.rowtime}} ,  and  should include both two 
> stream's rowtime attribute, {{o.rowtime between rowtime () and rowtime () + 
> 1}} should also not be supported.
> An row-time streams join will not be able to handle late data, because this 
> would mean in insert a row into a sorted order shift all other computations. 
> This would be too expensive to maintain. Therefore, we will throw an error if 
> a user tries to use an row-time stream join with late data handling.
> This issue includes:
> * Design of the DataStream operator to deal with stream join
> * Translation from Calcite's RelNode representation (LogicalJoin). 



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink pull request #4625: [FLINK-6233] [table] Support time-bounded stream i...

[xccui]

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

https://github.com/apache/flink/pull/4625#discussion_r143222605
  
--- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamWindowJoin.scala
 ---
@@ -184,4 +229,50 @@ class DataStreamWindowJoin(
 .returns(returnTypeInfo)
 }
   }
+
+  def createRowTimeInnerJoin(
+  leftDataStream: DataStream[CRow],
+  rightDataStream: DataStream[CRow],
+  returnTypeInfo: TypeInformation[CRow],
+  joinFunctionName: String,
+  joinFunctionCode: String,
+  leftKeys: Array[Int],
+  rightKeys: Array[Int]): DataStream[CRow] = {
+
+val rowTimeInnerJoinFunc = new RowTimeBoundedStreamInnerJoin(
+  leftLowerBound,
+  leftUpperBound,
+  allowedLateness = 0L,
+  leftSchema.typeInfo,
+  rightSchema.typeInfo,
+  joinFunctionName,
+  joinFunctionCode,
+  leftTimeIdx,
+  rightTimeIdx)
+
+if (!leftKeys.isEmpty) {
+  leftDataStream
+.connect(rightDataStream)
+.keyBy(leftKeys, rightKeys)
--- End diff --

Actually, I was quite confused about this condition `!leftKeys.isEmpty`. 
Since in `FlinkLogicalJoin.scala`, queries without equi-conditions are blocked, 
when will this condition be hold?


---

[jira] [Closed] (FLINK-4034) Dependency convergence on com.101tec:zkclient and com.esotericsoftware.kryo:kryo

[Piotr Nowojski (JIRA)]

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

Piotr Nowojski closed FLINK-4034.
-
Resolution: Duplicate

It seems like this work was abandoned. Closing as it is a subset of 
https://issues.apache.org/jira/browse/FLINK-7765

> Dependency convergence on com.101tec:zkclient and 
> com.esotericsoftware.kryo:kryo
> 
>
> Key: FLINK-4034
> URL: https://issues.apache.org/jira/browse/FLINK-4034
> Project: Flink
>  Issue Type: Bug
>  Components: Build System
>Affects Versions: 1.0.3
>Reporter: Vladislav Pernin
>
> If dependency convergence is enabled and asserted on Maven, projects using 
> Flink do not compile.
> Example :
> {code}
> Dependency convergence error for com.esotericsoftware.kryo:kryo:2.24.0 paths 
> to dependency are:
> +-groupidXXX:artifactidXXX:versionXXX
>   +-org.apache.flink:flink-java:1.0.3
> +-org.apache.flink:flink-core:1.0.3
>   +-com.esotericsoftware.kryo:kryo:2.24.0
> and
> +-groupidXXX:artifactidXXX:versionXXX
>   +-org.apache.flink:flink-streaming-java_2.11:1.0.3
> +-org.apache.flink:flink-runtime_2.11:1.0.3
>   +-com.twitter:chill_2.11:0.7.4
> +-com.twitter:chill-java:0.7.4
>   +-com.esotericsoftware.kryo:kryo:2.21
> and
> +-groupidXXX:artifactidXXX:versionXXX
>   +-org.apache.flink:flink-streaming-java_2.11:1.0.3
> +-org.apache.flink:flink-runtime_2.11:1.0.3
>   +-com.twitter:chill_2.11:0.7.4
> +-com.esotericsoftware.kryo:kryo:2.21
> {code}
>   
> {code}
> Dependency convergence error for com.101tec:zkclient:0.7 paths to dependency 
> are:
> +-groupidXXX:artifactidXXX:versionXXX
>   +-org.apache.flink:flink-connector-kafka-0.8_2.11:1.0.3
> +-org.apache.flink:flink-connector-kafka-base_2.11:1.0.3
>   +-com.101tec:zkclient:0.7
> and
> +-groupidXXX:artifactidXXX:versionXXX
>   +-org.apache.flink:flink-connector-kafka-0.8_2.11:1.0.3
> +-org.apache.kafka:kafka_2.11:0.8.2.2
>   +-com.101tec:zkclient:0.3
> {code}
> I cannot emit a pull request without knowing on which specifics versions you 
> rely.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-7774) Deserializers are not cleaned up when closing input streams

[ASF GitHub Bot (JIRA)]

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

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

GitHub user NicoK opened a pull request:

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

[FLINK-7774][network] fix not clearing deserializers on closing an input

## What is the purpose of the change

On cleanup of the `AbstractRecordReader`, `StreamInputProcessor`, and 
`StreamTwoInputProcessor`, the deserializers' current buffers are cleaned up 
but not their internal `spanningWrapper` and `nonSpanningWrapper` (via 
`RecordDeserializer#clear()`). This call should be added.

## Brief change log

- also clean up the internal `RecordDeserializer` state during cleanup of 
input channels

## Verifying this change

This change is a trivial rework / code cleanup without any test coverage.

## Does this pull request potentially affect one of the following parts:

  - Dependencies (does it add or upgrade a dependency): (no)
  - The public API, i.e., is any changed class annotated with 
`@Public(Evolving)`: (no)
  - The serializers: (no)
  - The runtime per-record code paths (performance sensitive): (no)
  - Anything that affects deployment or recovery: JobManager (and its 
components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)

## Documentation

  - Does this pull request introduce a new feature? (no)
  - If yes, how is the feature documented? (not applicable)



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

$ git pull https://github.com/NicoK/flink flink-7774

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

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


commit 341b3403d58337b87a8caffd87bc6a0f63292ea2
Author: Nico Kruber 
Date:   2017-09-04T15:21:52Z

[FLINK-7774][network] fix not clearing deserializers on closing an input




> Deserializers are not cleaned up when closing input streams
> ---
>
> Key: FLINK-7774
> URL: https://issues.apache.org/jira/browse/FLINK-7774
> Project: Flink
>  Issue Type: Bug
>  Components: Network
>Affects Versions: 1.4.0, 1.3.2
>Reporter: Nico Kruber
>Assignee: Nico Kruber
>
> On cleanup of the {{AbstractRecordReader}}, {{StreamInputProcessor}}, and 
> {{StreamTwoInputProcessor}}, the deserializers' current buffers are cleaned 
> up but not their internal {{spanningWrapper}} and {{nonSpanningWrapper}} via 
> {{RecordDeserializer#clear}}. This call should be added.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink pull request #4783: [FLINK-7774][network] fix not clearing deserialize...

[NicoK]

GitHub user NicoK opened a pull request:

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

[FLINK-7774][network] fix not clearing deserializers on closing an input

## What is the purpose of the change

On cleanup of the `AbstractRecordReader`, `StreamInputProcessor`, and 
`StreamTwoInputProcessor`, the deserializers' current buffers are cleaned up 
but not their internal `spanningWrapper` and `nonSpanningWrapper` (via 
`RecordDeserializer#clear()`). This call should be added.

## Brief change log

- also clean up the internal `RecordDeserializer` state during cleanup of 
input channels

## Verifying this change

This change is a trivial rework / code cleanup without any test coverage.

## Does this pull request potentially affect one of the following parts:

  - Dependencies (does it add or upgrade a dependency): (no)
  - The public API, i.e., is any changed class annotated with 
`@Public(Evolving)`: (no)
  - The serializers: (no)
  - The runtime per-record code paths (performance sensitive): (no)
  - Anything that affects deployment or recovery: JobManager (and its 
components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)

## Documentation

  - Does this pull request introduce a new feature? (no)
  - If yes, how is the feature documented? (not applicable)



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

$ git pull https://github.com/NicoK/flink flink-7774

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

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


commit 341b3403d58337b87a8caffd87bc6a0f63292ea2
Author: Nico Kruber 
Date:   2017-09-04T15:21:52Z

[FLINK-7774][network] fix not clearing deserializers on closing an input




---

[jira] [Commented] (FLINK-6233) Support rowtime inner equi-join between two streams in the SQL API

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/4625#discussion_r143211000
  
--- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/TimeBoundedStreamInnerJoin.scala
 ---
@@ -0,0 +1,410 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.join
+
+import java.util.{ArrayList, List => JList}
+
+import org.apache.flink.api.common.functions.FlatJoinFunction
+import org.apache.flink.api.common.state._
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.ListTypeInfo
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.codegen.Compiler
+import org.apache.flink.table.runtime.CRowWrappingCollector
+import org.apache.flink.table.runtime.types.CRow
+import org.apache.flink.table.util.Logging
+import org.apache.flink.types.Row
+import org.apache.flink.util.Collector
+
+/**
+  * A CoProcessFunction to execute time-bounded stream inner-join.
+  * Two kinds of time criteria:
+  * "L.time between R.time + X and R.time + Y" or "R.time between L.time - 
Y and L.time - X".
+  *
+  * @param leftLowerBound  the lower bound for the left stream (X in the 
criteria)
+  * @param leftUpperBound  the upper bound for the left stream (Y in the 
criteria)
+  * @param allowedLateness the lateness allowed for the two streams
+  * @param leftTypethe input type of left stream
+  * @param rightType   the input type of right stream
+  * @param genJoinFuncName the function code of other non-equi conditions
+  * @param genJoinFuncCode the function name of other non-equi conditions
+  *
+  */
+abstract class TimeBoundedStreamInnerJoin(
+private val leftLowerBound: Long,
+private val leftUpperBound: Long,
+private val allowedLateness: Long,
+private val leftType: TypeInformation[Row],
+private val rightType: TypeInformation[Row],
+private val genJoinFuncName: String,
+private val genJoinFuncCode: String,
+private val leftTimeIdx: Int,
+private val rightTimeIdx: Int)
+extends CoProcessFunction[CRow, CRow, CRow]
+with Compiler[FlatJoinFunction[Row, Row, Row]]
+with Logging {
+
+  private var cRowWrapper: CRowWrappingCollector = _
+
+  // the join function for other conditions
+  private var joinFunction: FlatJoinFunction[Row, Row, Row] = _
+
+  // cache to store rows from the left stream
+  private var leftCache: MapState[Long, JList[Row]] = _
+  // cache to store rows from the right stream
+  private var rightCache: MapState[Long, JList[Row]] = _
+
+  // state to record the timer on the left stream. 0 means no timer set
+  private var leftTimerState: ValueState[Long] = _
+  // state to record the timer on the right stream. 0 means no timer set
+  private var rightTimerState: ValueState[Long] = _
+
+  private val leftRelativeSize: Long = -leftLowerBound
+  private val rightRelativeSize: Long = leftUpperBound
+
+  private var leftExpirationTime: Long = 0L;
+  private var rightExpirationTime: Long = 0L;
+
+  protected var leftOperatorTime: Long = 0L
+  protected var rightOperatorTime: Long = 0L
+
+
+  // for delayed cleanup
+  private val cleanupDelay = (leftRelativeSize + rightRelativeSize) / 2
+
+  if (allowedLateness < 0) {
+throw new IllegalArgumentException("The allowed lateness must be 
non-negative.")
+  }
+
+  /**
+* Get the maximum interval between receiving a row and 

[GitHub] flink pull request #4625: [FLINK-6233] [table] Support time-bounded stream i...

[xccui]

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

https://github.com/apache/flink/pull/4625#discussion_r143211000
  
--- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/TimeBoundedStreamInnerJoin.scala
 ---
@@ -0,0 +1,410 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.join
+
+import java.util.{ArrayList, List => JList}
+
+import org.apache.flink.api.common.functions.FlatJoinFunction
+import org.apache.flink.api.common.state._
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.ListTypeInfo
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.codegen.Compiler
+import org.apache.flink.table.runtime.CRowWrappingCollector
+import org.apache.flink.table.runtime.types.CRow
+import org.apache.flink.table.util.Logging
+import org.apache.flink.types.Row
+import org.apache.flink.util.Collector
+
+/**
+  * A CoProcessFunction to execute time-bounded stream inner-join.
+  * Two kinds of time criteria:
+  * "L.time between R.time + X and R.time + Y" or "R.time between L.time - 
Y and L.time - X".
+  *
+  * @param leftLowerBound  the lower bound for the left stream (X in the 
criteria)
+  * @param leftUpperBound  the upper bound for the left stream (Y in the 
criteria)
+  * @param allowedLateness the lateness allowed for the two streams
+  * @param leftTypethe input type of left stream
+  * @param rightType   the input type of right stream
+  * @param genJoinFuncName the function code of other non-equi conditions
+  * @param genJoinFuncCode the function name of other non-equi conditions
+  *
+  */
+abstract class TimeBoundedStreamInnerJoin(
+private val leftLowerBound: Long,
+private val leftUpperBound: Long,
+private val allowedLateness: Long,
+private val leftType: TypeInformation[Row],
+private val rightType: TypeInformation[Row],
+private val genJoinFuncName: String,
+private val genJoinFuncCode: String,
+private val leftTimeIdx: Int,
+private val rightTimeIdx: Int)
+extends CoProcessFunction[CRow, CRow, CRow]
+with Compiler[FlatJoinFunction[Row, Row, Row]]
+with Logging {
+
+  private var cRowWrapper: CRowWrappingCollector = _
+
+  // the join function for other conditions
+  private var joinFunction: FlatJoinFunction[Row, Row, Row] = _
+
+  // cache to store rows from the left stream
+  private var leftCache: MapState[Long, JList[Row]] = _
+  // cache to store rows from the right stream
+  private var rightCache: MapState[Long, JList[Row]] = _
+
+  // state to record the timer on the left stream. 0 means no timer set
+  private var leftTimerState: ValueState[Long] = _
+  // state to record the timer on the right stream. 0 means no timer set
+  private var rightTimerState: ValueState[Long] = _
+
+  private val leftRelativeSize: Long = -leftLowerBound
+  private val rightRelativeSize: Long = leftUpperBound
+
+  private var leftExpirationTime: Long = 0L;
+  private var rightExpirationTime: Long = 0L;
+
+  protected var leftOperatorTime: Long = 0L
+  protected var rightOperatorTime: Long = 0L
+
+
+  // for delayed cleanup
+  private val cleanupDelay = (leftRelativeSize + rightRelativeSize) / 2
+
+  if (allowedLateness < 0) {
+throw new IllegalArgumentException("The allowed lateness must be 
non-negative.")
+  }
+
+  /**
+* Get the maximum interval between receiving a row and emitting it (as 
part of a joined result).
+* Only reasonable for row time join.
+*
+* @return the maximum delay for the outputs
+*/
+  def getMaxOutputDelay: Long = Math.max(leftRelativeSize, 

[GitHub] flink pull request #4625: [FLINK-6233] [table] Support time-bounded stream i...

[xccui]

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

https://github.com/apache/flink/pull/4625#discussion_r143208392
  
--- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/TimeBoundedStreamInnerJoin.scala
 ---
@@ -0,0 +1,410 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.join
+
+import java.util.{ArrayList, List => JList}
+
+import org.apache.flink.api.common.functions.FlatJoinFunction
+import org.apache.flink.api.common.state._
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.ListTypeInfo
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.codegen.Compiler
+import org.apache.flink.table.runtime.CRowWrappingCollector
+import org.apache.flink.table.runtime.types.CRow
+import org.apache.flink.table.util.Logging
+import org.apache.flink.types.Row
+import org.apache.flink.util.Collector
+
+/**
+  * A CoProcessFunction to execute time-bounded stream inner-join.
+  * Two kinds of time criteria:
+  * "L.time between R.time + X and R.time + Y" or "R.time between L.time - 
Y and L.time - X".
+  *
+  * @param leftLowerBound  the lower bound for the left stream (X in the 
criteria)
+  * @param leftUpperBound  the upper bound for the left stream (Y in the 
criteria)
+  * @param allowedLateness the lateness allowed for the two streams
+  * @param leftTypethe input type of left stream
+  * @param rightType   the input type of right stream
+  * @param genJoinFuncName the function code of other non-equi conditions
+  * @param genJoinFuncCode the function name of other non-equi conditions
+  *
+  */
+abstract class TimeBoundedStreamInnerJoin(
+private val leftLowerBound: Long,
+private val leftUpperBound: Long,
+private val allowedLateness: Long,
+private val leftType: TypeInformation[Row],
+private val rightType: TypeInformation[Row],
+private val genJoinFuncName: String,
+private val genJoinFuncCode: String,
+private val leftTimeIdx: Int,
+private val rightTimeIdx: Int)
+extends CoProcessFunction[CRow, CRow, CRow]
+with Compiler[FlatJoinFunction[Row, Row, Row]]
+with Logging {
+
+  private var cRowWrapper: CRowWrappingCollector = _
+
+  // the join function for other conditions
+  private var joinFunction: FlatJoinFunction[Row, Row, Row] = _
+
+  // cache to store rows from the left stream
+  private var leftCache: MapState[Long, JList[Row]] = _
+  // cache to store rows from the right stream
+  private var rightCache: MapState[Long, JList[Row]] = _
+
+  // state to record the timer on the left stream. 0 means no timer set
+  private var leftTimerState: ValueState[Long] = _
+  // state to record the timer on the right stream. 0 means no timer set
+  private var rightTimerState: ValueState[Long] = _
+
+  private val leftRelativeSize: Long = -leftLowerBound
+  private val rightRelativeSize: Long = leftUpperBound
+
+  private var leftExpirationTime: Long = 0L;
+  private var rightExpirationTime: Long = 0L;
+
+  protected var leftOperatorTime: Long = 0L
+  protected var rightOperatorTime: Long = 0L
+
+
+  // for delayed cleanup
+  private val cleanupDelay = (leftRelativeSize + rightRelativeSize) / 2
+
+  if (allowedLateness < 0) {
+throw new IllegalArgumentException("The allowed lateness must be 
non-negative.")
+  }
+
+  /**
+* Get the maximum interval between receiving a row and emitting it (as 
part of a joined result).
+* Only reasonable for row time join.
+*
+* @return the maximum delay for the outputs
+*/
+  def getMaxOutputDelay: Long = Math.max(leftRelativeSize, 

[jira] [Commented] (FLINK-6233) Support rowtime inner equi-join between two streams in the SQL API

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/4625#discussion_r143208392
  
--- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/TimeBoundedStreamInnerJoin.scala
 ---
@@ -0,0 +1,410 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.join
+
+import java.util.{ArrayList, List => JList}
+
+import org.apache.flink.api.common.functions.FlatJoinFunction
+import org.apache.flink.api.common.state._
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.ListTypeInfo
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.codegen.Compiler
+import org.apache.flink.table.runtime.CRowWrappingCollector
+import org.apache.flink.table.runtime.types.CRow
+import org.apache.flink.table.util.Logging
+import org.apache.flink.types.Row
+import org.apache.flink.util.Collector
+
+/**
+  * A CoProcessFunction to execute time-bounded stream inner-join.
+  * Two kinds of time criteria:
+  * "L.time between R.time + X and R.time + Y" or "R.time between L.time - 
Y and L.time - X".
+  *
+  * @param leftLowerBound  the lower bound for the left stream (X in the 
criteria)
+  * @param leftUpperBound  the upper bound for the left stream (Y in the 
criteria)
+  * @param allowedLateness the lateness allowed for the two streams
+  * @param leftTypethe input type of left stream
+  * @param rightType   the input type of right stream
+  * @param genJoinFuncName the function code of other non-equi conditions
+  * @param genJoinFuncCode the function name of other non-equi conditions
+  *
+  */
+abstract class TimeBoundedStreamInnerJoin(
+private val leftLowerBound: Long,
+private val leftUpperBound: Long,
+private val allowedLateness: Long,
+private val leftType: TypeInformation[Row],
+private val rightType: TypeInformation[Row],
+private val genJoinFuncName: String,
+private val genJoinFuncCode: String,
+private val leftTimeIdx: Int,
+private val rightTimeIdx: Int)
+extends CoProcessFunction[CRow, CRow, CRow]
+with Compiler[FlatJoinFunction[Row, Row, Row]]
+with Logging {
+
+  private var cRowWrapper: CRowWrappingCollector = _
+
+  // the join function for other conditions
+  private var joinFunction: FlatJoinFunction[Row, Row, Row] = _
+
+  // cache to store rows from the left stream
+  private var leftCache: MapState[Long, JList[Row]] = _
+  // cache to store rows from the right stream
+  private var rightCache: MapState[Long, JList[Row]] = _
+
+  // state to record the timer on the left stream. 0 means no timer set
+  private var leftTimerState: ValueState[Long] = _
+  // state to record the timer on the right stream. 0 means no timer set
+  private var rightTimerState: ValueState[Long] = _
+
+  private val leftRelativeSize: Long = -leftLowerBound
+  private val rightRelativeSize: Long = leftUpperBound
+
+  private var leftExpirationTime: Long = 0L;
+  private var rightExpirationTime: Long = 0L;
+
+  protected var leftOperatorTime: Long = 0L
+  protected var rightOperatorTime: Long = 0L
+
+
+  // for delayed cleanup
+  private val cleanupDelay = (leftRelativeSize + rightRelativeSize) / 2
+
+  if (allowedLateness < 0) {
+throw new IllegalArgumentException("The allowed lateness must be 
non-negative.")
+  }
+
+  /**
+* Get the maximum interval between receiving a row and 

[jira] [Commented] (FLINK-7072) Create RESTful cluster endpoint

[ASF GitHub Bot (JIRA)]

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

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

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4742
  
@tillrohrmann We got some test-failures for the CLIFrontend caused by us 
now using the ClusterClient for stop/cancel. This is purely a test construction 
issue, and I'm currently fixing them.


> Create RESTful cluster endpoint
> ---
>
> Key: FLINK-7072
> URL: https://issues.apache.org/jira/browse/FLINK-7072
> Project: Flink
>  Issue Type: Sub-task
>  Components: Distributed Coordination
>Reporter: Till Rohrmann
>Assignee: Chesnay Schepler
>  Labels: flip-6
> Fix For: 1.4.0
>
>
> In order to communicate with the cluster from the RESTful client, we have to 
> implement a RESTful cluster endpoint. The endpoint shall support the 
> following operations:
> * List jobs (GET): Get list of all running jobs on the cluster
> * Submit job (POST): Submit a job to the cluster (only supported in session 
> mode)
> * Get job status (GET): Get the status of an executed job (and maybe the 
> JobExecutionResult)
> * Lookup job leader (GET): Gets the JM leader for the given job
> This endpoint will run in session mode alongside the dispatcher/session 
> runner and forward calls to this component which maintains a view on all 
> currently executed jobs.
> In the per-job mode, the endpoint will return only the single running job and 
> the address of the JobManager alongside which it is running. Furthermore, it 
> won't accept job submissions.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink issue #4742: [FLINK-7072] [REST] Add Flip-6 client for submit/job/canc...

[zentol]

Github user zentol commented on the issue:

https://github.com/apache/flink/pull/4742
  
@tillrohrmann We got some test-failures for the CLIFrontend caused by us 
now using the ClusterClient for stop/cancel. This is purely a test construction 
issue, and I'm currently fixing them.


---

[jira] [Commented] (FLINK-7410) Use toString method to display operator names for UserDefinedFunction

[ASF GitHub Bot (JIRA)]

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

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

Github user hequn8128 commented on the issue:

https://github.com/apache/flink/pull/4624
  
Hi @fhueske , 
Thanks for your review, I have addressed all your comments and rebased the 
code to the master :)


> Use toString method to display operator names for UserDefinedFunction
> -
>
> Key: FLINK-7410
> URL: https://issues.apache.org/jira/browse/FLINK-7410
> Project: Flink
>  Issue Type: Improvement
>  Components: Table API & SQL
>Affects Versions: 1.4.0
>Reporter: Hequn Cheng
>Assignee: Hequn Cheng
>
> *Motivation*
> Operator names setted in table-api are used by visualization and logging, it 
> is import to make these names simple and readable. Currently, 
> UserDefinedFunction’s name contains class CanonicalName and md5 value making 
> the name too long and unfriendly to users. 
> As shown in the following example, 
> {quote}
> select: (a, b, c, 
> org$apache$flink$table$expressions$utils$RichFunc1$281f7e61ec5d8da894f5783e2e17a4f5(a)
>  AS _c3, 
> org$apache$flink$table$expressions$utils$RichFunc2$fb99077e565685ebc5f48b27edc14d98(c)
>  AS _c4)
> {quote}
> *Changes:*
>   
> Use {{toString}} method to display operator names for UserDefinedFunction. 
> The method will return class name by default. Users can also override the 
> method to return whatever he wants.
> What do you think [~fhueske] ?



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink issue #4624: [FLINK-7410] [table] Use toString method to display opera...

[hequn8128]

Github user hequn8128 commented on the issue:

https://github.com/apache/flink/pull/4624
  
Hi @fhueske , 
Thanks for your review, I have addressed all your comments and rebased the 
code to the master :)


---

[jira] [Commented] (FLINK-7770) Hide Queryable State behind a proxy.

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/4778#discussion_r143194039
  
--- Diff: 
flink-queryable-state/flink-queryable-state-java/src/test/java/itcases/HAQueryableStateITCaseFsBackend.java
 ---
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.flink.test.query;
+package itcases;
--- End diff --

Package name seems off, same for the other ones.


> Hide Queryable State behind a proxy.
> 
>
> Key: FLINK-7770
> URL: https://issues.apache.org/jira/browse/FLINK-7770
> Project: Flink
>  Issue Type: Improvement
>  Components: Queryable State
>Affects Versions: 1.4.0
>Reporter: Kostas Kloudas
>Assignee: Kostas Kloudas
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-7770) Hide Queryable State behind a proxy.

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/4778#discussion_r143193371
  
--- Diff: flink-queryable-state/flink-queryable-state-java/pom.xml ---
@@ -0,0 +1,149 @@
+
+
+http://maven.apache.org/POM/4.0.0;
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance;
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
http://maven.apache.org/maven-v4_0_0.xsd;>
+
+4.0.0
+
+
+org.apache.flink
+flink-queryable-state
+1.4-SNAPSHOT
+..
+
+
+
flink-queryable-state-java_${scala.binary.version}
+flink-queryable-state-java
+jar
+
+
+
+
+
+
+org.apache.flink
+flink-core
+${project.version}
+provided
+
+
+
+org.apache.flink
+
flink-streaming-java_${scala.binary.version}
+${project.version}
+provided
+
+
+
--- End diff --

I don't think we're supposed to have a Guava dependency here.


> Hide Queryable State behind a proxy.
> 
>
> Key: FLINK-7770
> URL: https://issues.apache.org/jira/browse/FLINK-7770
> Project: Flink
>  Issue Type: Improvement
>  Components: Queryable State
>Affects Versions: 1.4.0
>Reporter: Kostas Kloudas
>Assignee: Kostas Kloudas
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-7770) Hide Queryable State behind a proxy.

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/4778#discussion_r143193896
  
--- Diff: 
flink-queryable-state/flink-queryable-state-java/src/test/java/itcases/AbstractQueryableStateITCase.java
 ---
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.flink.test.query;
+package itcases;
--- End diff --

This package name seems a bit too short  


> Hide Queryable State behind a proxy.
> 
>
> Key: FLINK-7770
> URL: https://issues.apache.org/jira/browse/FLINK-7770
> Project: Flink
>  Issue Type: Improvement
>  Components: Queryable State
>Affects Versions: 1.4.0
>Reporter: Kostas Kloudas
>Assignee: Kostas Kloudas
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink pull request #4778: [FLINK-7770][FLINK-7769][Queryable State] Refactor...

[aljoscha]

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

https://github.com/apache/flink/pull/4778#discussion_r143194039
  
--- Diff: 
flink-queryable-state/flink-queryable-state-java/src/test/java/itcases/HAQueryableStateITCaseFsBackend.java
 ---
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.flink.test.query;
+package itcases;
--- End diff --

Package name seems off, same for the other ones.


---

[GitHub] flink pull request #4778: [FLINK-7770][FLINK-7769][Queryable State] Refactor...

[aljoscha]

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

https://github.com/apache/flink/pull/4778#discussion_r143193371
  
--- Diff: flink-queryable-state/flink-queryable-state-java/pom.xml ---
@@ -0,0 +1,149 @@
+
+
+http://maven.apache.org/POM/4.0.0;
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance;
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
http://maven.apache.org/maven-v4_0_0.xsd;>
+
+4.0.0
+
+
+org.apache.flink
+flink-queryable-state
+1.4-SNAPSHOT
+..
+
+
+
flink-queryable-state-java_${scala.binary.version}
+flink-queryable-state-java
+jar
+
+
+
+
+
+
+org.apache.flink
+flink-core
+${project.version}
+provided
+
+
+
+org.apache.flink
+
flink-streaming-java_${scala.binary.version}
+${project.version}
+provided
+
+
+
--- End diff --

I don't think we're supposed to have a Guava dependency here.


---

[GitHub] flink pull request #4778: [FLINK-7770][FLINK-7769][Queryable State] Refactor...

[aljoscha]

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

https://github.com/apache/flink/pull/4778#discussion_r143193896
  
--- Diff: 
flink-queryable-state/flink-queryable-state-java/src/test/java/itcases/AbstractQueryableStateITCase.java
 ---
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.flink.test.query;
+package itcases;
--- End diff --

This package name seems a bit too short 😉 


---

[jira] [Commented] (FLINK-7698) Join with null literals leads to NPE

[Xingcan Cui (JIRA)]

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

Xingcan Cui commented on FLINK-7698:


Hi [~twalthr], this problem has be addressed in 
[CALCITE-1860|https://issues.apache.org/jira/browse/CALCITE-1860]. It will be 
fixed in 1.14.

> Join with null literals leads to NPE
> 
>
> Key: FLINK-7698
> URL: https://issues.apache.org/jira/browse/FLINK-7698
> Project: Flink
>  Issue Type: Bug
>  Components: Table API & SQL
>Affects Versions: 1.4.0
>Reporter: Timo Walther
>Assignee: Xingcan Cui
>
> The following query fails:
> {code}
>   @Test
>   def testProcessTimeInnerJoin(): Unit = {
> val env = StreamExecutionEnvironment.getExecutionEnvironment
> val tEnv = TableEnvironment.getTableEnvironment(env)
> env.setStateBackend(getStateBackend)
> StreamITCase.clear
> env.setParallelism(1)
> val sqlQuery = "SELECT t2.a, t2.c, t1.c from T1 as t1 join T2 as t2 on 
> t1.a = t2.a and t1.nullField = t2.nullField and " +
>   "t1.proctime between t2.proctime - interval '5' second and t2.proctime 
> + interval '5' second"
> val data1 = new mutable.MutableList[(Int, Long, String)]
> data1.+=((1, 1L, "Hi1"))
> data1.+=((1, 2L, "Hi2"))
> data1.+=((1, 5L, "Hi3"))
> data1.+=((2, 7L, "Hi5"))
> data1.+=((1, 9L, "Hi6"))
> data1.+=((1, 8L, "Hi8"))
> data1.+=((1, 8L, "Hi8"))
> val data2 = new mutable.MutableList[(Int, Long, String)]
> data2.+=((1, 1L, "HiHi"))
> data2.+=((2, 2L, "HeHe"))
> val t1 = env.fromCollection(data1).toTable(tEnv, 'a, 'b, 'c, 
> 'proctime.proctime)
>   .select('a, 'b, 'c, 'proctime, Null(Types.LONG) as 'nullField)
> val t2 = env.fromCollection(data2).toTable(tEnv, 'a, 'b, 'c, 
> 'proctime.proctime)
>   .select('a, 'b, 'c, 'proctime, 12L as 'nullField)
> tEnv.registerTable("T1", t1)
> tEnv.registerTable("T2", t2)
> val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
> result.addSink(new StreamITCase.StringSink[Row])
> env.execute()
>   }
> {code}
> It leads to:
> {code}
> java.lang.NullPointerException
>   at org.apache.calcite.rex.RexUtil.gatherConstraint(RexUtil.java:437)
>   at org.apache.calcite.rex.RexUtil.gatherConstraints(RexUtil.java:399)
>   at org.apache.calcite.rex.RexUtil.predicateConstants(RexUtil.java:336)
>   at 
> org.apache.calcite.plan.RelOptPredicateList.of(RelOptPredicateList.java:144)
>   at 
> org.apache.calcite.rel.metadata.RelMdPredicates$JoinConditionBasedPredicateInference.inferPredicates(RelMdPredicates.java:654)
>   at 
> org.apache.calcite.rel.metadata.RelMdPredicates.getPredicates(RelMdPredicates.java:326)
>   at GeneratedMetadataHandler_Predicates.getPredicates_$(Unknown Source)
>   at GeneratedMetadataHandler_Predicates.getPredicates(Unknown Source)
>   at GeneratedMetadataHandler_Predicates.getPredicates_$(Unknown Source)
>   at GeneratedMetadataHandler_Predicates.getPredicates(Unknown Source)
>   at 
> org.apache.calcite.rel.metadata.RelMetadataQuery.getPulledUpPredicates(RelMetadataQuery.java:803)
>   at 
> org.apache.calcite.rel.rules.ReduceExpressionsRule$ProjectReduceExpressionsRule.onMatch(ReduceExpressionsRule.java:264)
>   at 
> org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:317)
>   at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:506)
>   at 
> org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:385)
>   at 
> org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:251)
>   at 
> org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:125)
>   at 
> org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:210)
>   at 
> org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:197)
>   at 
> org.apache.flink.table.api.TableEnvironment.runHepPlanner(TableEnvironment.scala:257)
>   at 
> org.apache.flink.table.api.StreamTableEnvironment.optimize(StreamTableEnvironment.scala:663)
>   at 
> org.apache.flink.table.api.StreamTableEnvironment.translate(StreamTableEnvironment.scala:728)
>   at 
> org.apache.flink.table.api.scala.StreamTableEnvironment.toAppendStream(StreamTableEnvironment.scala:219)
>   at 
> org.apache.flink.table.api.scala.StreamTableEnvironment.toAppendStream(StreamTableEnvironment.scala:195)
>   at 
> org.apache.flink.table.api.scala.TableConversions.toAppendStream(TableConversions.scala:121)
>   at 
> org.apache.flink.table.runtime.stream.sql.JoinITCase.testProcessTimeInnerJoin(JoinITCase.scala:67)
>   at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>   at 
> 

[jira] [Created] (FLINK-7774) Deserializers are not cleaned up when closing input streams

[Nico Kruber (JIRA)]

Nico Kruber created FLINK-7774:
--

 Summary: Deserializers are not cleaned up when closing input 
streams
 Key: FLINK-7774
 URL: https://issues.apache.org/jira/browse/FLINK-7774
 Project: Flink
  Issue Type: Bug
  Components: Network
Affects Versions: 1.3.2, 1.4.0
Reporter: Nico Kruber
Assignee: Nico Kruber


On cleanup of the {{AbstractRecordReader}}, {{StreamInputProcessor}}, and 
{{StreamTwoInputProcessor}}, the deserializers' current buffers are cleaned up 
but not their internal {{spanningWrapper}} and {{nonSpanningWrapper}} via 
{{RecordDeserializer#clear}}. This call should be added.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (FLINK-7773) Test instability in UtilsTest#testYarnFlinkResourceManagerJobManagerLostLeadership

[Nico Kruber (JIRA)]

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

Nico Kruber updated FLINK-7773:
---
Labels: test-stability  (was: )

> Test instability in 
> UtilsTest#testYarnFlinkResourceManagerJobManagerLostLeadership
> --
>
> Key: FLINK-7773
> URL: https://issues.apache.org/jira/browse/FLINK-7773
> Project: Flink
>  Issue Type: Bug
>  Components: Tests, YARN
>Affects Versions: 1.4.0
>Reporter: Nico Kruber
>Priority: Critical
>  Labels: test-stability
>
> {{UtilsTest#testYarnFlinkResourceManagerJobManagerLostLeadership}} may result 
> in the following exception (repeated run in IntelliJ until failure, but also 
> on Travis here: https://travis-ci.org/NicoK/flink/jobs/283696974 )
> {code}
> org.apache.flink.yarn.UtilsTest "Until Failure"
> org.mockito.exceptions.misusing.UnfinishedStubbingException: 
> Unfinished stubbing detected here:
> -> at org.apache.flink.yarn.UtilsTest$1.(UtilsTest.java:171)
> E.g. thenReturn() may be missing.
> Examples of correct stubbing:
> when(mock.isOk()).thenReturn(true);
> when(mock.isOk()).thenThrow(exception);
> doThrow(exception).when(mock).someVoidMethod();
> Hints:
>  1. missing thenReturn()
>  2. you are trying to stub a final method, you naughty developer!
>  3: you are stubbing the behaviour of another mock inside before 'thenReturn' 
> instruction if completed
>   at org.apache.flink.yarn.UtilsTest$1.(UtilsTest.java:179)
>   at 
> org.apache.flink.yarn.UtilsTest.testYarnFlinkResourceManagerJobManagerLostLeadership(UtilsTest.java:95)
>   at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>   at 
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
>   at 
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>   at java.lang.reflect.Method.invoke(Method.java:498)
>   at 
> org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
>   at 
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
>   at 
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
>   at 
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
>   at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:55)
>   at org.junit.rules.RunRules.evaluate(RunRules.java:20)
>   at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
>   at 
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
>   at 
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
>   at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
>   at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
>   at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
>   at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
>   at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
>   at 
> org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
>   at 
> org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
>   at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
>   at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
>   at 
> com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
>   at 
> com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:67)
>   at 
> com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
>   at 
> com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)
> {code}
> The incriminating code is this:
> {code}
> doAnswer(new Answer() {
>   @Override
>   public Object answer(InvocationOnMock invocation) throws Throwable {
>   Container container = (Container) invocation.getArguments()[0];
>   resourceManagerGateway.tell(new 
> NotifyResourceStarted(YarnFlinkResourceManager.extractResourceID(container)),
>   leader1Gateway);
>   return null;
>   }
> }).when(nodeManagerClient).startContainer(Matchers.any(Container.class), 
> Matchers.any(ContainerLaunchContext.class));
> {code}



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Created] (FLINK-7773) Test instability in UtilsTest#testYarnFlinkResourceManagerJobManagerLostLeadership

[Nico Kruber (JIRA)]

Nico Kruber created FLINK-7773:
--

 Summary: Test instability in 
UtilsTest#testYarnFlinkResourceManagerJobManagerLostLeadership
 Key: FLINK-7773
 URL: https://issues.apache.org/jira/browse/FLINK-7773
 Project: Flink
  Issue Type: Bug
  Components: Tests, YARN
Affects Versions: 1.4.0
Reporter: Nico Kruber
Priority: Critical


{{UtilsTest#testYarnFlinkResourceManagerJobManagerLostLeadership}} may result 
in the following exception (repeated run in IntelliJ until failure, but also on 
Travis here: https://travis-ci.org/NicoK/flink/jobs/283696974 )

{code}
org.apache.flink.yarn.UtilsTest "Until Failure"

org.mockito.exceptions.misusing.UnfinishedStubbingException: 
Unfinished stubbing detected here:
-> at org.apache.flink.yarn.UtilsTest$1.(UtilsTest.java:171)

E.g. thenReturn() may be missing.
Examples of correct stubbing:
when(mock.isOk()).thenReturn(true);
when(mock.isOk()).thenThrow(exception);
doThrow(exception).when(mock).someVoidMethod();
Hints:
 1. missing thenReturn()
 2. you are trying to stub a final method, you naughty developer!
 3: you are stubbing the behaviour of another mock inside before 'thenReturn' 
instruction if completed


at org.apache.flink.yarn.UtilsTest$1.(UtilsTest.java:179)
at 
org.apache.flink.yarn.UtilsTest.testYarnFlinkResourceManagerJobManagerLostLeadership(UtilsTest.java:95)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at 
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at 
org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
at 
org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at 
org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
at 
org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:55)
at org.junit.rules.RunRules.evaluate(RunRules.java:20)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
at 
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
at 
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
at 
org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
at 
org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
at 
com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
at 
com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:67)
at 
com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
at 
com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)
{code}

The incriminating code is this:
{code}
doAnswer(new Answer() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
Container container = (Container) invocation.getArguments()[0];
resourceManagerGateway.tell(new 
NotifyResourceStarted(YarnFlinkResourceManager.extractResourceID(container)),
leader1Gateway);
return null;
}
}).when(nodeManagerClient).startContainer(Matchers.any(Container.class), 
Matchers.any(ContainerLaunchContext.class));
{code}



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-7768) Load File Systems via Java Service abstraction

[ASF GitHub Bot (JIRA)]

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

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

Github user aljoscha commented on the issue:

https://github.com/apache/flink/pull/4781
  
The changes look good!  I had some comments about POMs, optional and 
shading but those got resolved "offline". I had one comment about a test name 
but feel free to merge.


> Load File Systems via Java Service abstraction
> --
>
> Key: FLINK-7768
> URL: https://issues.apache.org/jira/browse/FLINK-7768
> Project: Flink
>  Issue Type: Improvement
>  Components: Core
>Reporter: Stephan Ewen
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink issue #4781: [FLINK-7768] [core] Load File Systems via Java Service ab...

[aljoscha]

Github user aljoscha commented on the issue:

https://github.com/apache/flink/pull/4781
  
The changes look good! 👍 I had some comments about POMs, optional and 
shading but those got resolved "offline". I had one comment about a test name 
but feel free to merge.


---

[jira] [Updated] (FLINK-7772) Test instability in BlobCacheDeleteTest

[Nico Kruber (JIRA)]

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

Nico Kruber updated FLINK-7772:
---
Priority: Critical  (was: Major)

> Test instability in BlobCacheDeleteTest
> ---
>
> Key: FLINK-7772
> URL: https://issues.apache.org/jira/browse/FLINK-7772
> Project: Flink
>  Issue Type: Bug
>  Components: Network
>Affects Versions: 1.4.0
>Reporter: Nico Kruber
>Assignee: Nico Kruber
>Priority: Critical
>  Labels: test-stability
>
> {{BlobCacheDeleteTest}} did not account for the server executing the delete 
> call of a transient BLOB after acknowledging the request. This resulted in 
> the {{testDeleteTransientLocalFails*}} tests failing.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (FLINK-7772) Test instability in BlobCacheDeleteTest

[Nico Kruber (JIRA)]

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

Nico Kruber updated FLINK-7772:
---
Labels: test-stability  (was: )

> Test instability in BlobCacheDeleteTest
> ---
>
> Key: FLINK-7772
> URL: https://issues.apache.org/jira/browse/FLINK-7772
> Project: Flink
>  Issue Type: Bug
>  Components: Network
>Affects Versions: 1.4.0
>Reporter: Nico Kruber
>Assignee: Nico Kruber
>Priority: Critical
>  Labels: test-stability
>
> {{BlobCacheDeleteTest}} did not account for the server executing the delete 
> call of a transient BLOB after acknowledging the request. This resulted in 
> the {{testDeleteTransientLocalFails*}} tests failing.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLINK-7768) Load File Systems via Java Service abstraction

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/4781#discussion_r143174730
  
--- Diff: flink-dist/pom.xml ---
@@ -205,6 +205,32 @@ under the License.


 
+   
+
+   
+   org.apache.flink
+   flink-hadoop-fs
+   ${project.version}
+   
+   
+   org.apache.flink
+   
flink-shaded-hadoop2
+   
+   
+   
+   
+   
+   org.apache.flink
+   flink-mapr-fs
+   ${project.version}
--- End diff --

This works because flink-shaded-hadoop2 is a transitive optional 
dependency? We don't exclude it explicitly here.


> Load File Systems via Java Service abstraction
> --
>
> Key: FLINK-7768
> URL: https://issues.apache.org/jira/browse/FLINK-7768
> Project: Flink
>  Issue Type: Improvement
>  Components: Core
>Reporter: Stephan Ewen
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] flink pull request #4781: [FLINK-7768] [core] Load File Systems via Java Ser...

[aljoscha]

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

https://github.com/apache/flink/pull/4781#discussion_r143174730
  
--- Diff: flink-dist/pom.xml ---
@@ -205,6 +205,32 @@ under the License.


 
+   
+
+   
+   org.apache.flink
+   flink-hadoop-fs
+   ${project.version}
+   
+   
+   org.apache.flink
+   
flink-shaded-hadoop2
+   
+   
+   
+   
+   
+   org.apache.flink
+   flink-mapr-fs
+   ${project.version}
--- End diff --

This works because flink-shaded-hadoop2 is a transitive optional 
dependency? We don't exclude it explicitly here.


---

[jira] [Commented] (FLINK-7768) Load File Systems via Java Service abstraction

[ASF GitHub Bot (JIRA)]

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

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

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

https://github.com/apache/flink/pull/4781#discussion_r143171918
  
--- Diff: 
flink-filesystems/flink-mapr-fs/src/test/java/org/apache/flink/runtime/fs/maprfs/FileSystemAccessTest.java
 ---
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.fs.maprfs;
+
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * This test checks that the file system is properly accessible through the
+ * service loading abstraction.
+ */
+public class FileSystemAccessTest extends TestLogger {
--- End diff --

nit: have MapR in the test name


> Load File Systems via Java Service abstraction
> --
>
> Key: FLINK-7768
> URL: https://issues.apache.org/jira/browse/FLINK-7768
> Project: Flink
>  Issue Type: Improvement
>  Components: Core
>Reporter: Stephan Ewen
> Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)